import datetime import requests from loguru import logger import boto3 import os from botocore.config import Config import uuid import configparser # logger_file = '/root/develop/log/dotabot.log' logger_file = 'dotabot.log' logger.add(logger_file) def convert_seconds_to_hms(total_seconds): hours, remainder = divmod(total_seconds, 3600) minutes, seconds = divmod(remainder, 60) return hours, minutes, seconds def is_workday(): return datetime.datetime.today().weekday() < 4 def is_game_time(): # game time is workday 21:00 - 1:00, weekend 10:00 - 1:00 if is_workday(): # return datetime.datetime.now().hour >= 21 or datetime.datetime.now().hour < 1 return False else: return datetime.datetime.now().hour >= 12 or datetime.datetime.now().hour < 1 def heartbeat(): resp = requests.get('https://up.tunpok.com/api/push/BDb4MJWDVh?status=up&msg=OK&ping=') if not resp.ok: logger.error('fail to heartbeat, resp status: %s' % resp.status_code) def get_ranking(ranking_int): # (10-15: Herald, 20-25: Guardian, 30-35: Crusader, 40-45: Archon, 50-55: Legend, 60-65: Ancient, 70-75: Divine, 80-85: Immortal). if not ranking_int: return '' stars = ranking_int % 10 if ranking_int < 20: return '先锋 %s' % stars elif ranking_int < 30: return '卫士 %s' % stars elif ranking_int < 40: return '中军 %s' % stars elif ranking_int < 50: return '统帅 %s' % stars elif ranking_int < 60: return '传奇 %s' % stars elif ranking_int < 70: return '万古流芳 %s' % stars elif ranking_int < 80: return '超凡入圣 %s' % stars elif ranking_int < 90: return '不朽 %s' % stars else: return 'Unknown %s' % stars def shorten_digits(num): # if num like 12345, return 1.2w # if num like 1234, return 1.2k # if num < 1000, return num # if result ends with 0, remove it if num >= 10000: return '%.1fw' % (num / 10000) elif num >= 1000: return '%.1fk' % (num / 1000) else: return str(num) def get_hero_chinese_name(english_name): """将Dota英雄的英文名转换为中文名 Args: english_name (str): 英雄的英文名称 Returns: str: 英雄的中文名称，如果找不到对应的中文名则返回原英文名 """ hero_name_map = { "alchemist": "炼金术士", "axe": "斧王", "bristleback": "钢背兽", "centaur_warrunner": "半人马战行者", "chaos_knight": "混沌骑士", "dawnbreaker": "破晓辰星", "doom": "末日使者", "dragon_knight": "龙骑士", "earth_spirit": "大地之灵", "earthshaker": "撼地者", "elder_titan": "上古巨神", "huskar": "哈斯卡", "kunkka": "昆卡", "legion_commander": "军团指挥官", "lifestealer": "噬魂鬼", "mars": "玛尔斯", "night_stalker": "暗夜魔王", "ogre_magi": "食人魔魔法师", "omniknight": "全能骑士", "primal_beast": "兽", "pudge": "屠夫", "slardar": "斯拉达", "spirit_breaker": "裂魂人", "sven": "斯温", "tidehunter": "潮汐猎人", "timbersaw": "伐木机", "tiny": "小小", "treant_protector": "树精卫士", "tusk": "巨牙海民", "underlord": "孽主", "undying": "不朽尸王", "wraith_king": "冥魂大帝", "anti-mage": "敌法师", "arc_warden": "天穹守望者", "bloodseeker": "血魔", "bounty_hunter": "赏金猎人", "clinkz": "克林克兹", "drow_ranger": "卓尔游侠", "ember_spirit": "灰烬之灵", "faceless_void": "虚空假面", "gyrocopter": "矮人直升机", "hoodwink": "森海飞霞", "juggernaut": "主宰", "kez": "凯", "luna": "露娜", "medusa": "美杜莎", "meepo": "米波", "monkey_king": "齐天大圣", "morphling": "变体精灵", "naga_siren": "娜迦海妖", "phantom_assassin": "幻影刺客", "phantom_lancer": "幻影长矛手", "razor": "剃刀", "riki": "力丸", "shadow_fiend": "影魔", "slark": "斯拉克", "sniper": "狙击手", "spectre": "幽鬼", "templar_assassin": "圣堂刺客", "terrorblade": "恐怖利刃", "troll_warlord": "巨魔战将", "ursa": "熊战士", "viper": "冥界亚龙", "weaver": "编织者", "ancient_apparition": "远古冰魄", "crystal_maiden": "水晶室女", "death_prophet": "死亡先知", "disruptor": "干扰者", "enchantress": "魅惑魔女", "grimstroke": "天涯墨客", "jakiro": "杰奇洛", "keeper_of_the_light": "光之守卫", "leshrac": "拉席克", "lich": "巫妖", "lina": "莉娜", "lion": "莱恩", "muerta": "琼英碧灵", "nature's_prophet": "先知", "necrophos": "瘟疫法师", "oracle": "神谕者", "outworld_destroyer": "殁境神蚀者", "puck": "帕克", "pugna": "帕格纳", "queen_of_pain": "痛苦女王", "ringmaster": "百戏之王", "rubick": "拉比克", "shadow_demon": "暗影恶魔", "shadow_shaman": "暗影萨满", "silencer": "沉默术士", "skywrath_mage": "天怒法师", "storm_spirit": "风暴之灵", "tinker": "修补匠", "warlock": "术士", "witch_doctor": "巫医", "zeus": "宙斯", "abaddon": "亚巴顿", "bane": "祸乱之源", "batrider": "蝙蝠骑士", "beastmaster": "兽王", "brewmaster": "酿酒大师", "broodmother": "育母蜘蛛", "chen": "陈", "clockwerk": "发条技师", "dark_seer": "黑暗贤者", "dark_willow": "邪影芳灵", "dazzle": "戴泽", "enigma": "谜团", "invoker": "祈求者", "io": "艾欧", "lone_druid": "德鲁伊", "lycan": "狼人", "magnus": "马格纳斯", "marci": "玛西", "mirana": "米拉娜", "nyx_assassin": "司夜刺客", "pangolier": "石鳞剑士", "phoenix": "凤凰", "sand_king": "沙王", "snapfire": "电炎绝手", "techies": "工程师", "vengeful_spirit": "复仇之魂", "venomancer": "剧毒术士", "visage": "维萨吉", "void_spirit": "虚无之灵", "windranger": "风行者", "winter_wyvern": "寒冬飞龙" } # 将英文名转换为小写并去除空格，用作字典键 key = english_name.lower().replace(' ', '_') return hero_name_map.get(key, english_name) def upload_image(image_path, file_name=None): """ 将图片上传到 Cloudflare R2 存储桶并返回访问URL Args: image_path (str): 本地图片文件路径 Returns: str: 上传成功后的图片URL，失败则返回None """ try: # 读取配置文件 config = configparser.ConfigParser() # config_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'env.ini') config_path = './env.ini' if not os.path.exists(config_path): logger.error(f"配置文件不存在: {config_path}") return None config.read(config_path) # 获取Cloudflare R2配置 if 'cloudflare' not in config: logger.error("配置文件中缺少cloudflare部分") return None cf_config = config['cloudflare'] account_id = cf_config.get('account_id') access_key_id = cf_config.get('access_key_id') secret_access_key = cf_config.get('secret_access_key') bucket_name = cf_config.get('bucket_name', 'dotabot-images') region = cf_config.get('region', 'auto') custom_domain = cf_config.get('custom_domain', '') # 检查必要的配置 if not all([account_id, access_key_id, secret_access_key]): logger.error("缺少Cloudflare R2必要配置") return None # 创建S3客户端连接到Cloudflare R2 s3_client = boto3.client( 's3', endpoint_url=f'https://{account_id}.r2.cloudflarestorage.com', aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, region_name=region, config=Config( signature_version='s3v4', retries={ 'max_attempts': 3, 'mode': 'standard' } ) ) # 生成唯一的文件名 if not file_name: timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S') random_id = str(uuid.uuid4())[:8] file_extension = os.path.splitext(image_path)[1] object_key = f'dotabot/{timestamp}_{random_id}{file_extension}' else: object_key = f'dotabot/{file_name}' # 设置文件的Content-Type content_type = 'image/png' if file_extension.lower() == '.png' else 'image/jpeg' # 上传文件 with open(image_path, 'rb') as file_data: s3_client.upload_fileobj( file_data, bucket_name, object_key, ExtraArgs={ 'ContentType': content_type, 'CacheControl': 'max-age=2592000' # 30天缓存 } ) # 构建公共访问URL if custom_domain: image_url = f'https://{custom_domain}/{object_key}' else: image_url = f'https://{bucket_name}.{account_id}.r2.cloudflarestorage.com/{object_key}' logger.info(f"图片上传成功: {image_url}") return image_url except Exception as e: logger.error(f"上传图片失败: {str(e)}") return None