309 lines
9.8 KiB
Python
309 lines
9.8 KiB
Python
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 ''
|
||
ranking_int = int(ranking_int)
|
||
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:
|
||
file_extension = os.path.splitext(file_name)[1]
|
||
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 |