1. Cài đặt các thư viện cần thiết
Để bắt đầu làm việc với API của Huawei Cloud CDN, bạn cần cài đặt các module sau:
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple huaweicloudsdkcdn
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple requests
Lưu ý: Bạn cần có AK/SK để xác thực. Có thể lấy tại: Huawei Cloud IAM Console
2. Tải log CDN ngày hôm trước
Dưới đây là đoạn code hoàn chỉnh để tải log CDN của ngày hôm trước:
import os
import requests
import shutil
import gzip
import glob
import json
from datetime import datetime, timedelta
from huaweicloudsdkcore.auth.credentials import GlobalCredentials
from huaweicloudsdkcdn.v2.region.cdn_region import CdnRegion
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcdn.v2 import *
def lay_va_giai_nen_log(domain_name, start_timestamp, output_path, ngay_thu_muc):
"""Tải và giải nén log CDN cho một domain cụ thể"""
os.makedirs(output_path, exist_ok=True)
thu_muc_domain = os.path.join(output_path, domain_name)
os.makedirs(thu_muc_domain, exist_ok=True)
thu_muc_ngay = os.path.join(thu_muc_domain, str(ngay_thu_muc))
os.makedirs(thu_muc_ngay, exist_ok=True)
trang_hien_tai = 1
while True:
# Thay thế bằng AK/SK thực tế của bạn
access_key = 'YOUR_AK_HERE'
secret_key = 'YOUR_SK_HERE'
thong_tin_xac_thuc = GlobalCredentials(access_key, secret_key)
client = CdnClient.new_builder() \
.with_credentials(thong_tin_xac_thuc) \
.with_region(CdnRegion.value_of("cn-north-1")) \
.build()
try:
yeu_cau = ShowLogsRequest()
yeu_cau.domain_name = domain_name
yeu_cau.start_time = start_timestamp
yeu_cau.page_size = 60
yeu_cau.page_number = trang_hien_tai
yeu_cau.enterprise_project_id = "0"
phan_hoi = client.show_logs(yeu_cau)
du_lieu = json.loads(str(phan_hoi))
except exceptions.ClientRequestException as e:
print(f"Mã lỗi: {e.status_code}")
print(f"Request ID: {e.request_id}")
print(f"Mã lỗi chi tiết: {e.error_code}")
print(f"Thông báo lỗi: {e.error_msg}")
break
danh_sach_log = du_lieu.get('logs', [])
if not danh_sach_log:
break
for ban_ghi_log in danh_sach_log:
url_log = ban_ghi_log.get('link')
if url_log:
cac_thanh_phan = url_log.split('/')
phan_thu_7 = cac_thanh_phan[6]
ten_file = phan_thu_7.split('?')[0]
ten_log = os.path.basename(ten_file)
duong_dan_log = os.path.join(thu_muc_ngay, ten_log)
with requests.get(url_log, stream=True) as response:
with open(duong_dan_log, 'wb') as f:
shutil.copyfileobj(response.raw, f)
trang_hien_tai += 1
# Gộp và giải nén tất cả file trong một ngày cho cùng domain
duong_dan_log_gop = os.path.join(thu_muc_ngay, f'{domain_name}_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_gop, 'wb') as file_gop:
for file_gz in glob.glob(os.path.join(thu_muc_ngay, '*.gz')):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
os.remove(file_gz)
def chay_chuong_trinh():
danh_sach_domain = ["www.example.com"]
thoi_gian_hien_tai = datetime.now()
ngay_hom_qua = (thoi_gian_hien_tai - timedelta(days=1)).strftime('%Y-%m-%d')
ngay_hom_qua_dt = thoi_gian_hien_tai - timedelta(days=1)
nua_dem_hom_qua = datetime(year=ngay_hom_qua_dt.year, month=ngay_hom_qua_dt.month, day=ngay_hom_qua_dt.day)
start_timestamp = int(nua_dem_hom_qua.timestamp() * 1000)
duong_dan_luu = 'cdn_logs'
for domain in danh_sach_domain:
lay_va_giai_nen_log(domain, start_timestamp, duong_dan_luu, ngay_hom_qua)
if __name__ == "__main__":
chay_chuong_trinh()
3. Phân tách log theo khu vực (trong nước / quốc tế)
Huawei Cloud CDN lưu trữ log theo khu vực IP, bạn có thể tách riêng log truy cập từ IP trong nước và quốc tế:
import os
import requests
import shutil
import gzip
import glob
import json
from datetime import datetime, timedelta
from huaweicloudsdkcore.auth.credentials import GlobalCredentials
from huaweicloudsdkcdn.v2.region.cdn_region import CdnRegion
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcdn.v2 import *
def lay_va_giai_nen_log(domain_name, start_timestamp, output_path, ngay_thu_muc):
"""Tải log và phân tách theo khu vực địa lý"""
os.makedirs(output_path, exist_ok=True)
thu_muc_domain = os.path.join(output_path, domain_name)
os.makedirs(thu_muc_domain, exist_ok=True)
trang_hien_tai = 1
while True:
access_key = 'YOUR_AK_HERE'
secret_key = 'YOUR_SK_HERE'
thong_tin_xac_thuc = GlobalCredentials(access_key, secret_key)
client = CdnClient.new_builder() \
.with_credentials(thong_tin_xac_thuc) \
.with_region(CdnRegion.value_of("cn-north-1")) \
.build()
try:
yeu_cau = ShowLogsRequest()
yeu_cau.domain_name = domain_name
yeu_cau.start_time = start_timestamp
yeu_cau.page_size = 60
yeu_cau.page_number = trang_hien_tai
yeu_cau.enterprise_project_id = "0"
phan_hoi = client.show_logs(yeu_cau)
du_lieu = json.loads(str(phan_hoi))
except exceptions.ClientRequestException as e:
print(f"Mã lỗi: {e.status_code}")
print(f"Request ID: {e.request_id}")
print(f"Mã lỗi chi tiết: {e.error_code}")
print(f"Thông báo lỗi: {e.error_msg}")
break
danh_sach_log = du_lieu.get('logs', [])
if not danh_sach_log:
break
for ban_ghi_log in danh_sach_log:
url_log = ban_ghi_log.get('link')
if url_log:
cac_thanh_phan = url_log.split('/')
phan_thu_7 = cac_thanh_phan[6]
ten_file = phan_thu_7.split('?')[0]
ten_log = os.path.basename(ten_file)
duong_dan_log = os.path.join(thu_muc_domain, ten_log)
with requests.get(url_log, stream=True) as response:
with open(duong_dan_log, 'wb') as f:
shutil.copyfileobj(response.raw, f)
trang_hien_tai += 1
# Gộp tất cả log
duong_dan_log_tong = os.path.join(thu_muc_domain, f'{domain_name}_all_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_tong, 'wb') as file_gop:
for file_gz in sorted(glob.glob(os.path.join(thu_muc_domain, '*.gz'))):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
# Gộp log IP trong nước (cn)
duong_dan_log_cn = os.path.join(thu_muc_domain, f'{domain_name}_cn_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_cn, 'wb') as file_gop:
for file_gz in sorted(glob.glob(os.path.join(thu_muc_domain, '*cn.gz'))):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
os.remove(file_gz)
# Gộp log IP quốc tế (ov)
duong_dan_log_ov = os.path.join(thu_muc_domain, f'{domain_name}_ov_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_ov, 'wb') as file_gop:
for file_gz in sorted(glob.glob(os.path.join(thu_muc_domain, '*ov.gz'))):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
os.remove(file_gz)
def chay_chuong_trinh():
danh_sach_domain = ["www.example.com"]
thoi_gian_hien_tai = datetime.now()
ngay_hom_qua = (thoi_gian_hien_tai - timedelta(days=1)).strftime('%Y-%m-%d')
ngay_hom_qua_dt = thoi_gian_hien_tai - timedelta(days=1)
nua_dem_hom_qua = datetime(year=ngay_hom_qua_dt.year, month=ngay_hom_qua_dt.month, day=ngay_hom_qua_dt.day)
start_timestamp = int(nua_dem_hom_qua.timestamp() * 1000)
duong_dan_luu = 'cdn_logs'
for domain in danh_sach_domain:
lay_va_giai_nen_log(domain, start_timestamp, duong_dan_luu, ngay_hom_qua)
if __name__ == "__main__":
chay_chuong_trinh()
4. Tự động tải log nhiều ngày liên tiếp
Để tải log trong một khoảng thời gian dài (ví dụ 30 ngày), bạn có thể sử dụng vòng lặp:
import os
import requests
import shutil
import gzip
import glob
import json
from datetime import datetime, timedelta
from huaweicloudsdkcore.auth.credentials import GlobalCredentials
from huaweicloudsdkcdn.v2.region.cdn_region import CdnRegion
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcdn.v2 import *
def lay_va_giai_nen_log(domain_name, start_timestamp, output_path, ngay_thu_muc):
"""Tải log cho nhiều ngày liên tiếp"""
os.makedirs(output_path, exist_ok=True)
thu_muc_domain = os.path.join(output_path, domain_name)
os.makedirs(thu_muc_domain, exist_ok=True)
trang_hien_tai = 1
while True:
access_key = 'YOUR_AK_HERE'
secret_key = 'YOUR_SK_HERE'
thong_tin_xac_thuc = GlobalCredentials(access_key, secret_key)
client = CdnClient.new_builder() \
.with_credentials(thong_tin_xac_thuc) \
.with_region(CdnRegion.value_of("cn-north-1")) \
.build()
try:
yeu_cau = ShowLogsRequest()
yeu_cau.domain_name = domain_name
yeu_cau.start_time = start_timestamp
yeu_cau.page_size = 60
yeu_cau.page_number = trang_hien_tai
yeu_cau.enterprise_project_id = "0"
phan_hoi = client.show_logs(yeu_cau)
du_lieu = json.loads(str(phan_hoi))
except exceptions.ClientRequestException as e:
print(f"Mã lỗi: {e.status_code}")
print(f"Request ID: {e.request_id}")
print(f"Mã lỗi chi tiết: {e.error_code}")
print(f"Thông báo lỗi: {e.error_msg}")
break
danh_sach_log = du_lieu.get('logs', [])
if not danh_sach_log:
break
for ban_ghi_log in danh_sach_log:
url_log = ban_ghi_log.get('link')
if url_log:
cac_thanh_phan = url_log.split('/')
phan_thu_7 = cac_thanh_phan[6]
ten_file = phan_thu_7.split('?')[0]
ten_log = os.path.basename(ten_file)
duong_dan_log = os.path.join(thu_muc_domain, ten_log)
with requests.get(url_log, stream=True) as response:
with open(duong_dan_log, 'wb') as f:
shutil.copyfileobj(response.raw, f)
trang_hien_tai += 1
# Gộp tất cả log
duong_dan_log_tong = os.path.join(thu_muc_domain, f'{domain_name}_all_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_tong, 'wb') as file_gop:
for file_gz in glob.glob(os.path.join(thu_muc_domain, '*.gz')):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
# Gộp log IP trong nước
duong_dan_log_cn = os.path.join(thu_muc_domain, f'{domain_name}_cn_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_cn, 'wb') as file_gop:
for file_gz in glob.glob(os.path.join(thu_muc_domain, '*cn.gz')):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
os.remove(file_gz)
# Gộp log IP quốc tế
duong_dan_log_ov = os.path.join(thu_muc_domain, f'{domain_name}_ov_{str(ngay_thu_muc)}.log')
with open(duong_dan_log_ov, 'wb') as file_gop:
for file_gz in glob.glob(os.path.join(thu_muc_domain, '*ov.gz')):
with gzip.open(file_gz, 'rb') as f_in:
shutil.copyfileobj(f_in, file_gop)
os.remove(file_gz)
def chay_chuong_trinh():
danh_sach_domain = ["www.example.com", "api.example.com"]
duong_dan_luu = 'cdn_logs'
# Ngày bắt đầu
ngay_bat_dau = datetime(2024, 4, 8)
# Số ngày cần tải
so_ngay = 10
for ngay in range(so_ngay):
ngay_hien_tai = ngay_bat_dau + timedelta(days=ngay)
ngay_thu_muc = ngay_hien_tai.strftime("%Y-%m-%d")
start_timestamp = int(ngay_hien_tai.timestamp() * 1000)
for domain in danh_sach_domain:
lay_va_giai_nen_log(domain, start_timestamp, duong_dan_luu, ngay_thu_muc)
if __name__ == "__main__":
chay_chuong_trinh()