File size: 2,290 Bytes
65d3d39 522e726 65d3d39 8786c0f 65d3d39 8786c0f 65d3d39 8786c0f 65d3d39 71baa3b 65d3d39 8786c0f 65d3d39 522e726 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
# coding: utf-8
import asyncio
import os
import requests
from os.path import join
# from uvicorn import run
from fastapi import FastAPI
from fastapi.responses import Response
from loguru import logger
from __init__ import DIR_PATH, api_token
from utils import push_msg, read_logfile, get_cookie_from_pi
from api import start_daemon_scheduler, kill_daemon_scheduler
schedule_loop = asyncio.new_event_loop()
app = FastAPI()
@app.get("/")
def home():
html_content = open("index.html").read()
return Response(content=html_content, status_code=200)
# 开启定时任务
@app.get("/run")
async def do_run(token: str):
if token != api_token:
return {"code": 422}
await get_cookie_from_pi()
start_daemon_scheduler(schedule_loop)
res = {"code": 0, "data": "start scheduler"}
return res
# 关闭定时任务
@app.get("/kill")
def do_kill(token: str):
if token != api_token:
return {"code": 422}
asyncio.run_coroutine_threadsafe(kill_daemon_scheduler(), loop=schedule_loop)
res = {"code": 0, "data": "kill scheduler"}
return res
# 查询日志
@app.get("/log")
async def do_log(token: str, name: str):
if token != api_token:
return {"code": 422}
log_file = join(DIR_PATH, f"log/{name}.log")
data = await read_logfile(log_file)
res = {"code": 200, "data": data}
return res
# 列出当前目录所有文件
@app.get("/path")
def do_path(token: str):
if token != api_token:
return {"code": 422}
data = []
for root, _, files in os.walk(DIR_PATH):
for file in files:
data.append(join(root, file))
res = {"code": 0, "data": data}
return res
# 测试连通性
@app.get("/ip")
def do_ip(token: str):
if token != api_token:
return {"code": 422}
ip = requests.get("https://checkip.amazonaws.com").text.strip()
logger.debug(f"checking ip: {ip}")
res = {"code": 0, "ip": ip}
return res
# 测试推送消息
@app.get("/push")
def do_push(token: str):
if token != api_token:
return {"code": 422}
data = push_msg("send from huggingface docker", "hello world")
return {"code": 0, "data": data}
## 不要开两个实例
# if __name__ == "__main__":
# run("app:app", host="0.0.0.0", port=8000, reload=False)
|