m2pool_docs/m2pool_backend_app/app.py

import schedule
import time
import asyncio
import sys
from datetime import datetime, timedelta

from public.times import Times
from src.hashrate import HashRate
from src.report import Report, ReportEnx
from src.confirm import Confirm
from src.distribution import Distribution
from src.blanace import Balance, DGBBlance
from src.clear import ClearDBData
from src.notice import Notice


def main():
    # 获取命令行参数
    if len(sys.argv) < 3:
        raise Exception("请提供方法和币种参数")
    
    method = sys.argv[1]
    methods = ["hashrate", "report", "clear", "distribution", "confirm", "balance", "stats", "notice"]
    
    if method not in methods:
        raise Exception(f"暂不支持{method}方法")
    
    coin = sys.argv[2]
    coins = ["nexa", "mona", "grs", "dgbq", "dgbs", "dgbo", "rxd", "enx", "alph"]
    
    if coin not in coins:
        raise Exception(f"暂不支持{coin}")
    
    # hashrate 任务
    if method == "hashrate":
        hashrate = HashRate(coin)
        
        def hashrate_job():
            asyncio.run(hashrate_task(hashrate))
        
        # 每5分钟的第30秒执行
        schedule.every().hour.at(":00:30").do(hashrate_job)
        schedule.every().hour.at(":05:30").do(hashrate_job)
        schedule.every().hour.at(":10:30").do(hashrate_job)
        schedule.every().hour.at(":15:30").do(hashrate_job)
        schedule.every().hour.at(":20:30").do(hashrate_job)
        schedule.every().hour.at(":25:30").do(hashrate_job)
        schedule.every().hour.at(":30:30").do(hashrate_job)
        schedule.every().hour.at(":35:30").do(hashrate_job)
        schedule.every().hour.at(":40:30").do(hashrate_job)
        schedule.every().hour.at(":45:30").do(hashrate_job)
        schedule.every().hour.at(":50:30").do(hashrate_job)
        schedule.every().hour.at(":55:30").do(hashrate_job)
    
    # report 任务
    elif method == "report":
        if coin == "enx":
            report = ReportEnx(coin)
        else:
            report = Report(coin)
        
        interval = 60  # 秒
        if coin == "rxd":
            interval = 300
        elif coin == "nexa":
            interval = 120
        
        def report_job():
            report.main()
        
        schedule.every(interval).seconds.do(report_job)
    
    # confirm 任务
    elif method == "confirm":
        interval = 60  # 秒
        if coin == "rxd":
            interval = 300
        elif coin == "nexa":
            interval = 120
        
        confirm = Confirm(coin)
        
        def confirm_job():
            confirm.main()
        
        schedule.every(interval).seconds.do(confirm_job)
    
    # distribution 任务
    elif method == "distribution":
        distribution = Distribution(coin)
        
        now_ts = datetime.now()
        last_ts = now_ts - timedelta(hours=24)
        
        ymd_now = Times.utc_time(now_ts.isoformat())
        ymd_last = Times.utc_time(last_ts.isoformat())
        
        end_time = ymd_now.split(" ")[0] + " 00:00:00"
        start_time = ymd_last.split(" ")[0] + " 00:00:00"
        
        distribution.main(start_time, end_time)
    
    # balance 任务
    elif method == "balance":
        special_coins = ["dgbo", "dgbs", "dgbq"]
        
        if coin in special_coins:
            balance = DGBBlance(coin)
        else:
            balance = Balance(coin)
        
        hour = 4
        if coin in ["rxd", "alph"]:
            hour = 9
        
        async def balance_task_async(balance_obj):
            count = 0
            last_height = await balance_obj.node.getblockcount()
            
            while count < 36:  # 最多执行 36 次 (6小时)
                enable = await balance_obj.query_now_height(last_height)
                if enable:
                    result = await balance_obj.main()
                    if not result:
                        print(f"{coin}转账已完成")
                        return  # 成功执行后退出循环
                
                print(f"等待中... (已等待 {count * 10} 分钟)")
                await asyncio.sleep(1000 * 60 * 10)  # 休眠 10 分钟
                count += 1
            
            print("等待超时，任务结束！")
        
        def balance_job():
            asyncio.run(balance_task_async(balance))
        
        schedule.every().day.at(f"{hour:02d}:10:00").do(balance_job)
    
    # clear 任务
    elif method == "clear":
        clear = ClearDBData(coin)
        
        try:
            clear.clear_shares_db(72)
            print("sharesdb:ok")
            clear.clear_hashrate_db()
            print("hashratedb:ok")
        except Exception as err:
            print(err)
        finally:
            sys.exit(0)
    
    # notice 任务
    elif method == "notice":
        notice = Notice(coin)
        
        def notice_job():
            notice.main()
        
        schedule.every().day.at("09:30:00").do(notice_job)
    
    # 运行调度器
    while True:
        schedule.run_pending()
        time.sleep(1)


async def hashrate_task(hashrate):
    """hashrate 异步任务"""
    ymd_now = Times.utc_time(datetime.now().isoformat())
    ymd = ymd_now.split(":")
    end_time = ymd[0] + ":" + ymd[1] + ":00"
    
    await hashrate.insert_hashrate_miners_table(end_time)
    
    current_minute = datetime.now().minute
    data = await hashrate.query_hashrate_miners_accepts(end_time)
    
    if current_minute == 0 or current_minute == 30:
        await hashrate.insert_mhs(data)
        await hashrate.insert_mhs_real(data)
    else:
        await hashrate.insert_mhs_real(data)


if __name__ == "__main__":
    main()