import requests
|
from datetime import datetime
|
import time
|
import random
|
import os
|
import json
|
from auto_media_publisher.config.conf_base import BASE_DIR,DAJIALA_JSON_FILE_NAME,DAJIALA_CATEGORY_DICT,DAJIALA_URL_PREFIX
|
from auto_media_publisher.utils.utils_database import dajiala_insert_or_update_top15_json
|
from auto_media_publisher.utils.logger import get_logger
|
|
class DajialaFetcher():
|
def __init__(self):
|
super().__init__()
|
self.base_url = DAJIALA_URL_PREFIX
|
self.category_dict = DAJIALA_CATEGORY_DICT
|
self.output_json_file = os.path.join(BASE_DIR,DAJIALA_JSON_FILE_NAME)
|
self.logger = get_logger(self.__class__.__name__,"admin")
|
|
|
def fetch(self,top_type):
|
url = f"{self.base_url}{top_type}"
|
headers = {
|
"Accept":"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
"Accept-language":"zh-CN,zh;q=0.9",
|
"Accept-encoding":"gzip, deflate, br, zstd",
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36",
|
"Host":"www.dajiala.com",
|
"Pragma":"no-cache",
|
"Sec-ch-ua":"\"Google Chrome\";v=\"131\", \"Chromium\";v=\"131\", \"Not_A Brand\";v=\"24\"",
|
"Sec-ch-ua-mobile":"?0",
|
"Sec-ch-ua-platform":"\"Windows\"",
|
"Sec-fetch-dest":"document",
|
"Sec-fetch-mode":"navigate",
|
"Sec-fetch-site":"none",
|
"Sec-fetch-user":"?1",
|
"Upgrade-insecure-requests":"1"
|
}
|
|
try:
|
response = requests.get(url, headers=headers, timeout=10)
|
response.raise_for_status()
|
return response.json()
|
except requests.RequestException as e:
|
self.logger.error(f"[Top15Fetcher] Failed to fetch topType {top_type}: {e}")
|
return None
|
|
def store_to_jsonl_and_database(self, json_data,top_type):
|
if not json_data or json_data.get("code") != 0:
|
self.logger.warning(f"[Top15Fetcher] No data or invalid response for top_type={top_type}")
|
return
|
|
data_list = json_data["data"]["data"]
|
print(f"[Top15Fetcher] Saving {len(data_list)} items to JSONL for top_type {top_type}")
|
|
# 构建去重索引（内存）
|
seen_urls = set()
|
if os.path.exists(self.output_json_file):
|
with open(self.output_json_file, "r", encoding="utf-8") as f:
|
for line in f:
|
try:
|
obj = json.loads(line)
|
seen_urls.add(obj["url"])
|
except Exception:
|
continue
|
|
# 追加写入新数据（不重复）
|
with open(self.output_json_file, "a", encoding="utf-8") as f:
|
for item in data_list:
|
if item["url"] in seen_urls:
|
continue
|
item["type"] = top_type
|
item["type_name"] = self.category_dict.get(top_type, "未知")
|
item["updatetime"] = datetime.now().isoformat()
|
f.write(json.dumps(item, ensure_ascii=False) + "\n")
|
# store into database
|
dajiala_insert_or_update_top15_json(item)
|
|
self.logger.info(f"[Top15Fetcher] Stored new unique items to {self.output_json_file} and database")
|
|
|
|
def run(self):
|
for top_type in range(1, 27): # 1 to 26
|
data = self.fetch(top_type)
|
if data:
|
self.store_to_jsonl_and_database(data,top_type)
|
time.sleep(random.uniform(10, 15))
|
|
