douyin-spider/api.py

# -*- encoding: utf-8 -*-
'''
@File    :   api.py
@Time    :   2023年06月19日 20:20:58 星期一
@Author  :   erma0
@Version :   0.1
@Link    :   https://erma0.cn
@Desc    :   抖音爬虫封装API
'''

import os
import shutil
import stat
import threading
import time
from copy import deepcopy
from enum import Enum
from typing import List, Union

import ujson as json
import uvicorn
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, UJSONResponse
from pydantic import BaseModel
from uvicorn.config import LOGGING_CONFIG

from browser import Browser
from spider import Douyin

date_fmt = "%Y-%m-%d %H:%M:%S"
LOGGING_CONFIG["formatters"]["access"]["fmt"] = '%(asctime)s %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s'
LOGGING_CONFIG["formatters"]["default"]["fmt"] = "%(asctime)s %(levelprefix)s %(message)s"
LOGGING_CONFIG["formatters"]["default"]["datefmt"] = date_fmt
LOGGING_CONFIG["formatters"]["access"]["datefmt"] = date_fmt

title: str = '抖音爬虫API'
version: str = "0.1.0"
update = "20230619"
author = "erma0"
web = "https://douyin.erma0.cn/"
github = "https://github.com/erma0/douyin"
doc = "https://douyin.erma0.cn/docs"

description: str = """
### ❤️开源不易，欢迎star⭐

- 支持采集账号主页作品、喜欢作品、音乐原声作品、搜索作品、关注列表、粉丝列表、合集作品、单个作品

### 📢声明

> 本仓库为学习`playwright`爬虫、命令行调用`Aria2`及`FastAPI/AMIS/Eel`实现`WEBUI`的案例，仅用于测试和学习研究，禁止用于商业用途或任何非法用途。

> 任何用户直接或间接使用、传播本仓库内容时责任自负，本仓库的贡献者不对该等行为产生的任何后果负责。

> 如果相关方认为该项目的代码可能涉嫌侵犯其权利，请及时通知删除相关代码。
"""
contact = {"author": author, "url": github}
tags_metadata = [
    {
        "name": "同步",
        "description": "耗时短的接口，直接返回结果",
    },
    {
        "name": "异步",
        "description": "耗时长的接口，无法直接返回结果，返回一个轮询接口",
    },
]

edge = None
running_ls = []
download_path = './下载_API'

app = FastAPI(
    title=title,
    description=description,
    version=version,
    openapi_tags=tags_metadata,
    contact=contact,
)
# 允许跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


class DouyinAPI(Douyin):

    def _append_awemes(self, aweme_list: List[dict]):
        super()._append_awemes(aweme_list)
        with open(f"{download_path}/{self.type}_{self.id}.json", 'w', encoding='utf-8') as f:
            json.dump({'code': 1, 'num': len(self.results)}, f, ensure_ascii=False)

    def _append_users(self, user_list: List[dict]):
        super()._append_users(user_list)
        with open(f"{download_path}/{self.type}_{self.id}.json", 'w', encoding='utf-8') as f:
            json.dump({'code': 1, 'num': len(self.results)}, f, ensure_ascii=False)


def running(a: DouyinAPI, target: str):
    # a.aria2_conf = f"{download_path}/{a.type}_{a.id}.txt"
    a.run()
    with open(f"{download_path}/{a.type}_{a.id}.json", 'w', encoding='utf-8') as f:
        json.dump({
            'code': 0,
            'num': len(a.results),
            'data': a.results,
        }, f, ensure_ascii=False)
    running_ls.remove(target)


class API(BaseModel):
    Version: str = version
    Update: str = update
    Web: str = web
    GitHub: str = github
    Doc: str = doc
    Time: str = time.ctime()


class DataVideo(BaseModel):
    id: str
    desc: str
    download_addr: Union[str, List[str]]
    time: int = None
    digg_count: int = None
    share_count: int = None
    collect_count: int = None
    comment_count: int = None
    diggCount: int = None
    shareCount: int = None
    collectCount: int = None
    commentCount: int = None
    liveWatchCount: int = None
    music_title: str = None
    music_url: str = None
    cover: str = None
    tags: List[dict] = None


class DataUser(BaseModel):
    nickname: str
    sec_uid: str
    uid: str = None
    signature: str = None
    avatar: str = None
    short_id: str = None
    unique_id: str = None
    unique_id_modify_time: int = None
    aweme_count: int = None
    favoriting_count: int = None
    follower_count: int = None
    following_count: int = None
    constellation: int = None
    create_time: int = None
    enterprise_verify_reason: str = None
    is_gov_media_vip: bool = None
    total_favorited: int = None
    share_qrcode_uri: str = None


class DataSync(BaseModel):
    code: int = 0  # 0 已完成；1 正在运行
    num: int = 0
    data: List[Union[DataVideo, DataUser]] = None


class DataAsync(BaseModel):
    code: int = 0  # 0 成功投递；1 正在运行；
    url: str


class TypeAsync(str, Enum):
    post = 'post'
    like = 'like'
    favorite = 'favorite'
    music = 'music'
    search = 'search'
    collection = 'collection'
    follow = 'follow'
    fans = 'fans'


@app.get(
    "/",
    response_class=UJSONResponse,
    response_model=API,
)
def api(req: Request):
    return {'Web': req.base_url._url, 'Doc': f'{req.base_url._url}docs'}


@app.get(
    "/api/video",
    response_class=UJSONResponse,
    response_model=DataVideo,
    tags=['同步'],
    response_model_exclude_unset=True,
    response_model_exclude_defaults=True,
)
def get_video(url: str):
    start_browser()
    a = DouyinAPI(context=edge.context, url=url, type='video', down_path=download_path)
    a.run()
    return deepcopy(a.results[0])


@app.get(
    "/api/user",
    response_class=UJSONResponse,
    response_model=DataUser,
    tags=['同步'],
    response_model_exclude_unset=True,
    response_model_exclude_defaults=True,
)
def get_user(id: str):
    start_browser()
    a = DouyinAPI(context=edge.context, url=id, type='id', down_path=download_path)
    a.run()
    return deepcopy(a.results[0])


@app.get("/api/info", response_class=UJSONResponse, tags=['同步'])
def get_info(url: str):
    start_browser()
    a = DouyinAPI(context=edge.context, url=url, down_path=download_path)
    a.has_more = False
    a.run()
    return deepcopy(a.info)


@app.get(
    "/api/{type_async}",
    response_class=UJSONResponse,
    response_model=DataAsync,
    tags=['异步'],
    status_code=201,
)
def start_async(type_async: TypeAsync, url: str, background_tasks: BackgroundTasks, req: Request):
    start_browser()
    target = f"{type_async.value}_{url.strip()}"
    a = DouyinAPI(context=edge.context, url=url, type=type_async.value, down_path=download_path)
    if target in running_ls:
        code = 1
        # return RedirectResponse(url=f"{req.base_url._url}/api/{type_async.value}/{a.id}", status_code=303)
    else:
        code = 0
        running_ls.append(target)
        background_tasks.add_task(running, a, target)
    return {'code': code, 'url': f"{req.base_url._url}api/{type_async.value}/{a.id}"}


@app.get(
    "/api/{type_async}/{id}",
    response_class=UJSONResponse,
    response_model=DataSync,
    tags=['同步'],
    response_model_exclude_unset=True,
    response_model_exclude_defaults=True,
)
def async_result(type_async: TypeAsync, id: str, down: bool = False):
    suffix = 'txt' if down else 'json'
    file = f"{download_path}/{type_async.value}_{id}.{suffix}"
    if not os.path.exists(file):
        raise HTTPException(status_code=404, detail="目标不存在")
    return FileResponse(file)


@app.get("/t", response_class=UJSONResponse)
def tt():
    print('start')
    time.sleep(3)
    print('end')
    return {"url": "type"}


@app.get("/init")
def start_browser():
    global edge
    if not isinstance(edge, Browser):
        edge = Browser(headless=False)


# 删除只读文件夹
def remove_readonly(func, path, _):
    os.chmod(path, stat.S_IWRITE)
    func(path)


# 每10分钟清理一次下载文件夹，删除1小时以前的记录
def clean_download_path():
    root = download_path
    if not os.path.exists(root): os.makedirs(root)
    while True:
        now = time.time()
        for file in os.listdir(root):
            file_path = os.path.join(root, file)
            if now - os.path.getmtime(file_path) > 60 * 60 * 1:
                if os.path.isfile(file_path):
                    os.remove(file_path)
                else:
                    shutil.rmtree(file_path, onerror=remove_readonly)
        time.sleep(60 * 10)


# 启动后执行
@app.on_event("startup")
def startup():
    threading.Thread(target=clean_download_path, daemon=True).start()


if __name__ == '__main__':
    # uvicorn api:app --host '0.0.0.0' --port 567 --reload
    # uvicorn.run("api:app", host='0.0.0.0', port=567, reload=True, limit_concurrency=5, use_colors=True)
    uvicorn.run("api:app", host='0.0.0.0', port=567, limit_concurrency=5, use_colors=True)