数据采集监控与告警：错误重试、日志分析与自动化运维

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging

# 配置日志记录，便于后续的日志分析与监控
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 代理设置（参考亿牛云爬虫代理 www.16yun.cn）
proxy_host = "proxy.16yun.cn"   # 代理域名
proxy_port = "8080"                # 代理端口
proxy_user = "16YUN"                # 代理用户名
proxy_pass = "16IP"                # 代理密码

# 组装代理URL，格式为：http://username:password@host:port
proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"

proxies = {
    "http": proxy_url,
    "https": proxy_url
}

# 设置请求头，包括Cookie和User-Agent（可根据需要修改cookie值）
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36",
    "Cookie": "your_cookie_here"
}

# 定义错误重试策略
retry_strategy = Retry(
    total=3,                          # 总共重试3次
    status_forcelist=[429, 500, 502, 503, 504],  # 针对这些状态码进行重试
    backoff_factor=1                  # 重试延时因子
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)

def fetch_company_info(company_name):
    """
    根据公司名称查询天眼查上的企业信息，
    包括统一社会信用代码、注册资本、成立日期、股东及法律风险等信息
    """
    # 构造查询URL（此处为示例，实际接口需根据目标网站页面结构编写解析逻辑）
    url = f"https://www.tianyancha.com/search?key={company_name}"
    
    try:
        response = session.get(url, headers=headers, proxies=proxies, timeout=10)
        response.raise_for_status()  # 检查请求是否成功
        logger.info(f"成功获取数据，状态码: {response.status_code}")
        
        # 以下为解析逻辑示例，可使用BeautifulSoup解析HTML内容：
        # from bs4 import BeautifulSoup
        # soup = BeautifulSoup(response.text, 'html.parser')
        # 提取统一社会信用代码、注册资本、成立日期、股东、法律风险等信息
        
        # 模拟返回数据
        company_info = {
            "统一社会信用代码": "1234567890ABCDEFG",
            "注册资本": "5000万元",
            "成立日期": "2010-01-01",
            "股东": ["股东A", "股东B"],
            "法律风险": "无"
        }
        return company_info
    except Exception as e:
        logger.error(f"请求失败: {e}")
        # 此处可加入自动告警逻辑，如发送邮件、短信或调用运维接口进行干预
        # send_alert(e)
        return None

if __name__ == "__main__":
    company_name = "示例公司"
    info = fetch_company_info(company_name)
    if info:
        logger.info("获取的企业信息:")
        for key, value in info.items():
            logger.info(f"{key}: {value}")
    else:
        logger.error("未能获取企业信息")