import osimport jsonimport timeimport loggingimport requestsimport scheduleimport sqlite3from datetime import datetime, timedeltafrom typing import Optionalfrom configparser import ConfigParserfrom tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
# ==================== 配置和日志设置 ==================== #
# 初始化配置解析器config = ConfigParser()config.read('config.ini')
# 企业微信配置WECHAT_CONFIG = config['WeChat']CORP_ID = WECHAT_CONFIG.get('CORP_ID')AGENT_ID = WECHAT_CONFIG.getint('AGENT_ID')SECRET = WECHAT_CONFIG.get('SECRET')TO_USER = WECHAT_CONFIG.get('TO_USER')
# 日报来源配置REPORT_CONFIG = config['Report']DB_PATH = REPORT_CONFIG.get('DB_PATH', 'daily_report.db')FILE_PATH = REPORT_CONFIG.get('FILE_PATH', 'daily_report.txt')API_URL = REPORT_CONFIG.get('API_URL')
# 调度配置SCHEDULE_CONFIG = config['Schedule']SEND_TIME = SCHEDULE_CONFIG.get('SEND_TIME', '09:00')
# 日志配置logging.basicConfig(    level=logging.INFO,    format='%(asctime)s [%(levelname)s] %(message)s',    handlers=[        logging.FileHandler('daily_report.log', encoding='utf-8'),        logging.StreamHandler()    ])
# ==================== 企业微信客户端类 ==================== #
class WeChatClient:    def __init__(self, corp_id: str, secret: str, agent_id: int):        self.corp_id = corp_id        self.secret = secret        self.agent_id = agent_id        self.access_token: Optional[str] = None        self.token_expires_at: datetime = datetime.now()
    @retry(stop=stop_after_attempt(3), wait=wait_fixed(2), retry=retry_if_exception_type(requests.RequestException))    def get_access_token(self) -> str:        if self.access_token and datetime.now() < self.token_expires_at:            return self.access_token
        url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'        params = {'corpid': self.corp_id, 'corpsecret': self.secret}        response = requests.get(url, params=params)        response.raise_for_status()        data = response.json()
        if data.get('errcode') != 0:            raise ValueError(f"获取access_token失败: {data.get('errmsg')}")
        self.access_token = data.get('access_token')        expires_in = data.get('expires_in', 7200)        self.token_expires_at = datetime.now() + timedelta(seconds=expires_in - 300)  # 提前5分钟过期
        logging.info("成功获取新的access_token")        return self.access_token
    @retry(stop=stop_after_attempt(3), wait=wait_fixed(2), retry=retry_if_exception_type(requests.RequestException))    def send_text_message(self, to_user: str, content: str) -> None:        token = self.get_access_token()        url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}'        headers = {'Content-Type': 'application/json'}        payload = {            'touser': to_user,            'msgtype': 'text',            'agentid': self.agent_id,            'text': {'content': content},            'safe': 0        }        response = requests.post(url, headers=headers, json=payload)        response.raise_for_status()        data = response.json()
        if data.get('errcode') != 0:            raise ValueError(f"消息发送失败: {data.get('errmsg')}")
        logging.info(f"消息发送成功，接收人: {to_user}")
# ==================== 日报获取器类 ==================== #
class ReportFetcher:    @staticmethod    def fetch_from_db(db_path: str) -> Optional[str]:        today = datetime.now().strftime('%Y-%m-%d')        try:            with sqlite3.connect(db_path) as conn:                cursor = conn.cursor()                cursor.execute("SELECT content FROM reports WHERE date = ?", (today,))                result = cursor.fetchone()                if result:                    logging.info("成功从数据库获取日报内容")                    return result[0]                else:                    logging.warning("数据库中无今日日报内容")                    return None        except sqlite3.Error as e:            logging.error(f"数据库操作失败: {e}")            return None
    @staticmethod    def fetch_from_file(file_path: str) -> Optional[str]:        try:            with open(file_path, 'r', encoding='utf-8') as file:                content = file.read()                logging.info("成功从文件获取日报内容")                return content        except FileNotFoundError:            logging.warning(f"未找到日报文件: {file_path}")            return None        except IOError as e:            logging.error(f"读取日报文件失败: {e}")            return None
    @staticmethod    @retry(stop=stop_after_attempt(3), wait=wait_fixed(2), retry=retry_if_exception_type(requests.RequestException))    def fetch_from_api(api_url: str) -> Optional[str]:        try:            response = requests.get(api_url)            response.raise_for_status()            logging.info("成功从API获取日报内容")            return response.text        except requests.RequestException as e:            logging.error(f"从API获取日报内容失败: {e}")            return None
    @classmethod    def get_daily_report(cls) -> str:        content = cls.fetch_from_db(DB_PATH)        if content:            return content
        content = cls.fetch_from_file(FILE_PATH)        if content:            return content
        if API_URL:            content = cls.fetch_from_api(API_URL)            if content:                return content
        logging.error("无法获取日报内容，将发送默认提示信息")        return "今日没有可用的日报内容。"
# ==================== 主函数和调度 ==================== #
def send_daily_report():    wechat_client = WeChatClient(CORP_ID, SECRET, AGENT_ID)    report_content = ReportFetcher.get_daily_report()    try:        wechat_client.send_text_message(TO_USER, report_content)    except Exception as e:        logging.error(f"发送日报失败: {e}")
def main():    schedule.every().day.at(SEND_TIME).do(send_daily_report)    logging.info(f"任务已启动，将在每日 {SEND_TIME} 发送日报")    while True:        schedule.run_pending()        time.sleep(60)  # 每分钟检查一次任务
if __name__ == "__main__":    main()

[WeChat]CORP_ID = your_corp_idAGENT_ID = your_agent_idSECRET = your_secretTO_USER = user1|user2
[Report]DB_PATH = daily_report.dbFILE_PATH = daily_report.txtAPI_URL = https://api.example.com/daily_report
[Schedule]SEND_TIME = 09:00

pip install requests schedule tenacity