LongArticlesJob/applications/api/gzh_api.py

import json
import requests
from tenacity import retry
from requests.exceptions import RequestException
from typing import Optional, Dict

from applications import log
from applications.utils import request_retry

retry_desc = request_retry(retry_times=3, min_retry_delay=2, max_retry_delay=60)
base_url = "http://crawler-cn.aiddit.com/crawler/wei_xin"
headers = {"Content-Type": "application/json"}

def send_post_request(url, data):
    try:
        response = requests.post(url, headers=headers, data=data, timeout=60)
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        print(f"API请求失败: {e}")
    except json.JSONDecodeError as e:
        print(f"响应解析失败: {e}")
    return None


@retry(**retry_desc)
def search_articles_in_gzh(title: str, page: str = "1") -> Optional[Dict]:
    url = f"{base_url}/keyword"
    payload = json.dumps(
        {
            "keyword": title,
            "cursor": page
        }
    )
    return send_post_request(url, data=payload)


@retry(**retry_desc)
def get_gzh_article_detail(content_link: str, is_count: bool = False, is_cache: bool =True) -> Optional[Dict]:
    url = f"{base_url}/detail"
    payload = json.dumps(
        {
            "content_link": content_link,
            "is_count": is_count,
            "is_cache": is_cache,
            "is_ad": False
        }
    )
    return send_post_request(url, data=payload)


@retry(**retry_desc)
def get_gzh_account_article_list(gh_id: str, index: str = None) -> Optional[Dict]:
    url = f"{base_url}/blogger"
    payload = json.dumps(
        {
            "account_id": gh_id,
            "cursor": index
        }
    )
    return send_post_request(url, data=payload)


@retry(**retry_desc)
def get_gzh_account_detail(content_link: str) -> Optional[Dict]:
    url = f"{base_url}/account_info"
    payload = json.dumps(
        {
            "content_link": content_link
        }
    )
    return send_post_request(url, data=payload)


@retry(**retry_desc)
def get_gzh_recommend_articles(content_link: str) -> Optional[Dict]:
    url = f"{base_url}/recommend"
    payload = json.dumps(
        {
            "content_link": content_link
        }
    )
    return send_post_request(url, data=payload)


@retry(**retry_desc)
def get_gzh_recommend_articles_v2(content_link: str) -> Optional[Dict]:
    url = "http://datapi.top/wxapi/relatedarticle"
    payload = {
        'url': content_link,
        'token': '401e4d3c85068bb5'
    }
    response =  send_post_request(url, data=payload)
    log(
        task="article_association_crawler",
        function="get_recommend_articles_v2",
        message="获取推荐链接，付费接口",
        data={"content_link": content_link, "response": response},
    )
    return response