server.py

import json
from fastapi import FastAPI,Query,WebSocket
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import life
from fastapi import FastAPI, Request,HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
import logging

import uvicorn
from fastapi.middleware.cors import CORSMiddleware
import os
import time
from datetime import datetime
from typing import Dict, Any


def require_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


CALENDAR_API_PASSWORD = require_env("REALRPG_CALENDAR_API_PASSWORD")


app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],  # 允许跨域的源
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有方法 (GET, POST 等)
    allow_headers=["*"],  # 允许所有请求头
)

logging.basicConfig(level=logging.INFO)

# 获取访问者ip
@app.get("/get-ip")
async def get_ip(request: Request):
    client_ip = request.client.host
    return JSONResponse(content={"ip": client_ip})

# 工作日志接口
# 定义API接口，返回所有life表中的数据
@app.get("/calendar/data/")
async def fetch_data(name: str = Query(..., description="姓名")):
    try:
        work = life.Life()
        if name == "lyy":
            work = life.Life("Data/life_lyy.db")
        data = await work.query_all_data()
        return data
    except Exception as e:
        print(e)
        return False
    
@app.get("/calendar/today_data/")
async def fetch_data(timestamp: float = Query(..., description="时间戳"),name: str = Query(..., description="姓名")):
    """获取指定时间戳当天的数据"""
    try:
        work = life.Life()
        if name == "lyy":
            work = life.Life("Data/life_lyy.db")
            # 调用 query_today_data 方法，并传入查询参数中的时间戳
        data = await work.query_today_data(timestamp=timestamp)
        return data  # 返回查询结果
    except Exception as e:
        print(e)
        return {"error": str(e)}  # 返回错误信息
   
class RequestData(BaseModel):
    type: str
    content: str
    status: bool
    password: str
@app.post("/calendar/new_record")
async def receive_data(data: RequestData,name: str = Query(..., description="姓名")):
    try:
        work = life.Life()
        if name == "lyy":
            work = life.Life("Data/life_lyy.db")
        work_type = data.type
        work_content = data.content
        work_status = data.status
        password = data.password    
        if password == CALENDAR_API_PASSWORD:
            if await work.insert_life_entry(work_type,work_content):
                return {"message": "Data received successfully", "data": data}
        else:
            return {"message": "密码错误"}
    except Exception as e:
        print(e)
    return {"message": "未知错误"}
    
# 监控网站访问
# 创建日志目录
log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'request_logs')
if not os.path.exists(log_dir):
    os.makedirs(log_dir)

async def handle_request(request: Request) -> Dict[str, Any]:
    """处理和记录请求的详细信息"""
    # 获取请求体
    body = {}
    try:
        body = await request.json()
    except:
        try:
            body = await request.form()
            body = dict(body)
        except:
            try:
                body_data = await request.body()
                body = {"raw": body_data.decode('utf-8', errors='ignore')}
            except:
                pass

    # 收集请求详细信息
    request_info = {
        "timestamp": datetime.now().isoformat(),
        "method": request.method,
        "url": {
            "full": str(request.url),
            "path": request.url.path,
            "protocol": "https" if request.url.scheme == "https" else "http",
            "host": request.url.hostname,
        },
        "client": {
            "ip": request.client.host if request.client else None,
            "xForwardedFor": request.headers.get('X-Forwarded-For'),
            "xRealIp": request.headers.get('X-Real-Ip'),
        },
        "headers": dict(request.headers),
        "queryParams": dict(request.query_params),
        "body": body,
        "fingerprint": {
            "userAgent": request.headers.get('User-Agent'),
            "acceptLanguage": request.headers.get('Accept-Language'),
            "acceptEncoding": request.headers.get('Accept-Encoding'),
            "acceptHeaders": request.headers.get('Accept'),
            "dnt": request.headers.get('DNT'),
            "connection": request.headers.get('Connection'),
            "secFetchSite": request.headers.get('Sec-Fetch-Site'),
            "secFetchMode": request.headers.get('Sec-Fetch-Mode'),
            "secFetchDest": request.headers.get('Sec-Fetch-Dest'),
            "secChUa": request.headers.get('Sec-CH-UA'),
            "secChUaPlatform": request.headers.get('Sec-CH-UA-Platform'),
            "secChUaMobile": request.headers.get('Sec-CH-UA-Mobile'),
        },
    }

    # 将请求信息保存到文件
    timestamp = int(time.time() * 1000)
    file_prefix = request.url.path.replace('/', '_').lstrip('_') if request.url.path != '/' else 'root'
    filename = f"{file_prefix}-{timestamp}.json"
    file_path = os.path.join(log_dir, filename)

    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump(request_info, f, ensure_ascii=False, indent=2)

    print(f"请求详情已保存至: {file_path}")
    return request_info

@app.get("/wallet_check_in/info")
@app.post("/wallet_check_in/info")
@app.put("/wallet_check_in/info")
@app.delete("/wallet_check_in/info")
async def wallet_check_in(request: Request):
    print('检测到钱包检查请求!')
    print('钱包公钥:', request.query_params.get('walletPubkey'))
    
    await handle_request(request)
    
    return {
        "code": 200,
        "msg": "success",
        "data": {
            "config": [
                {
                    "raffleTimesTarget": 0,
                    "pointTarget": 900,
                    "dayNeed": 7,
                    "finish": 0
                },
                {
                    "raffleTimesTarget": 0,
                    "pointTarget": 2000,
                    "dayNeed": 15,
                    "finish": 0
                },
                {
                    "raffleTimesTarget": 0,
                    "pointTarget": 4500,
                    "dayNeed": 30,
                    "finish": 0
                },
                {
                    "raffleTimesTarget": 0,
                    "pointTarget": 12000,
                    "dayNeed": 60,
                    "finish": 0
                }
            ],
            "boostingTitle": "Limited time - 4x points",
            "uiType": 1,
            "emojiId": 0,
            "pointSingle": 100,
            "pointTotal": 0,
            "currentTimes": 0,
            "needTimes": 7
        },
        "total": ''
    }

@app.get("/wallet_check_in/upload")
@app.post("/wallet_check_in/upload")
@app.put("/wallet_check_in/upload")
@app.delete("/wallet_check_in/upload")
async def wallet_upload(request: Request):
    print('检测到钱包检查请求!')
    print('钱包公钥:', request.query_params.get('walletPubkey'))
    
    await handle_request(request)
    
    return {
        "code": 200,
        "msg": "success",
        "data": '',
        "total": ''
    }

@app.get("/track")
@app.post("/track")
@app.put("/track")
@app.delete("/track")
async def track(request: Request):
    print('检测到钱包检查请求!')
    print('钱包公钥:', request.query_params.get('walletPubkey'))
    
    await handle_request(request)
    
    return {
        "code": 200,
        "msg": "success",
        "data": '',
        "total": ''
    }

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def catch_all(path: str, request: Request):
    # 处理请求并记录
    await handle_request(request)
    
    # 返回通用响应
    timestamp = datetime.now().isoformat()
    file_prefix = path.replace('/', '_') if path else 'root'
    request_id = f"{file_prefix}-{int(time.time() * 1000)}"
    
    return {
        "message": "请求已成功记录和分析",
        "requestId": request_id,
        "timestamp": timestamp
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=9655)