群晖运行Python服务器

这篇具有很好参考价值的文章主要介绍了群晖运行Python服务器。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

import asyncio
import glob
import os
import sys
from concurrent.futures.thread import ThreadPoolExecutor
import aiofiles
import aiohttp
import psutil
import requests
from quart import request, Quart, json, send_file, redirect

process_pid = set()
process_pid.add(os.getpid())


def clean():
    """
    清理之前的所有python进程与浏览器进程
    """
    py_process_name = os.path.basename(sys.executable)
    py_code_file_name = os.path.basename(__file__)
    for p in psutil.process_iter():
        # kill another me
        process_cmdline = p.cmdline()
        if not p.cmdline():
            continue
        if py_process_name in str(process_cmdline) and py_code_file_name in str(process_cmdline):
            if p.pid not in process_pid:
                p.kill()
                print("已有启动中的实例，结束： " + str(p))
            else:
                print("当前python进程：" + str(p))


app = Quart(__name__)

cache_ip = dict()
thread_pool = ThreadPoolExecutor(max_workers=10)
http_client = requests.session()

MAIN_DIR = "/volume2"
APK_DATA_DIR = "/volume2/wj/1"
APK_DIR = "/volume2/wj/2"


def list_file_with_time_sorted(d):
    list_dir_abs = [os.path.join(d, file_name) for file_name in os.listdir(d)]
    return sorted(filter(os.path.isfile, list_dir_abs), key=os.path.getmtime, reverse=True)


def list_dir_with_time_sorted(d):
    list_dir_abs = [os.path.join(d, file_name) for file_name in os.listdir(d)]
    return sorted(filter(os.path.isdir, list_dir_abs), key=os.path.getmtime, reverse=True)


async def update_frpc_online_loop():
    print("预备刷新在线设备")


async def get_ip_loc_pc(ip: str) -> str:
    # return "获取IP地址失败：" + ip
    try:
        if ip in cache_ip:
            return cache_ip[ip]
        # http://whois.pconline.com.cn/ipJson.jsp?ip=36.62.138.185&json=true
        url = f"http://whois.pconline.com.cn/ipJson.jsp?ip={ip}&json=true"
        ret = await asyncio.wait_for(asyncio.get_event_loop().run_in_executor(thread_pool, http_client.get, url),
                                     timeout=2)
        addr = ret.json().get("addr")
        cache_ip.update({ip: addr})
        return addr
    except asyncio.exceptions.TimeoutError or asyncio.exceptions.CancelledError:
        return "IP_ADDRESS_TIME_OUT_2S"
    except:
        return "获取IP失败"


@app.route('/test')
async def _t():
    return "<br/>".join(list_file_with_time_sorted(MAIN_DIR))


@app.route('/app_ensuredir', methods=['GET', 'POST'])
async def _ensure_dir():
    form = await request.form
    dir_ensure = form.get("dir")
    dir_target = os.path.join(MAIN_DIR, dir_ensure)
    print("=", dir_target)
    if not os.path.exists(dir_target):
        os.makedirs(dir_target)
        os.system(f"chmod -R 766 \"{dir_target}\"")
    return dir_ensure


@app.route('/app_one_ver', methods=['GET', 'POST'])
async def _list_one_version():
    """
    单个App备份 返回版本选择
    :return:
    """
    form = await request.form
    meid = request.headers.get("meid")
    aid = form.get("aid")
    ret = ""
    app_data_dir = f"{APK_DATA_DIR}/{meid}"
    count = 0
    for app_backup_file in list_file_with_time_sorted(app_data_dir):
        # 最多返回10个，如果有多余则删除
        if aid in app_backup_file:
            if count < 10:
                size_in_mb = os.path.getsize(os.path.join(app_data_dir, app_backup_file)) / (1024 * 1024)
                ret += "{} 大小:{}Mb".format(os.path.basename(app_backup_file), int(size_in_mb) + 1)
                ret += "\t"
                count += 1
            else:
                break
    print(ret)
    return ret


@app.route('/app_all_ver', methods=['GET', 'POST'])
async def _list_all_version():
    """
    所有App备份 返回版本选择
    :return:
    """
    count = 0
    ret = ""
    for app_backup_file in list_dir_with_time_sorted(APK_DIR):
        if count > 10:
            break
        app_count = len(list_file_with_time_sorted(app_backup_file))
        ret += "{}个 {}".format(app_count, os.path.basename(app_backup_file)) + "\t"
        count += 1
    return ret


@app.route('/app_all_list', methods=['GET', 'POST'])
async def _list_summary():
    """
    所有App列表 记录在 summary.txt
    :return:
    """
    tag = (await request.form).get("tag")
    app_data_dir = f"{APK_DIR}/{tag}/summary.txt"
    print(app_data_dir)
    if os.path.exists(app_data_dir):
        async with aiofiles.open(app_data_dir, mode="r") as f:
            return await f.read()
    else:
        return ""


@app.route('/ff/<name>/<type_>', methods=['GET', 'POST'])
async def oss_(name, type_):
    try:
        path = glob.glob(f"{name}/*.{type_}")[0]
        print("发送文件:" + path, request.remote_addr, request.headers.get("User-Agent", ""),
              await get_ip_loc_pc(request.remote_addr))
        name = os.path.basename(path)
        return await send_file(path, attachment_filename=name, as_attachment=True)
    except:
        return f"{name}/*.{type_} not found"


clean()
loop = asyncio.get_event_loop()
# loop.create_task(update_frpc_online_loop())
app.run("0.0.0.0", port=9999, debug=True, loop=loop)

安装python3
到应用套件搜索安装即可

 wget  https://bootstrap.pypa.io/get-pip.py
 sudo python3 get-pip.py
 sudo pip install quart
 sudo pip install psutil
 sudo pip install aiofiles
 sudo pip install aiohttp
 sudo pip install requests