1
import time
2
import sqlite3
3
from pydantic import BaseModel
4
from fastapi import FastAPI, Depends, Request
5
from fastapi.middleware.cors import CORSMiddleware
6
from slowapi import Limiter, _rate_limit_exceeded_handler
7
from slowapi.util import get_remote_address
8
from slowapi.errors import RateLimitExceeded
9

10
DATABASE_PATH = "reactions.db"
11

12
def get_db_connection():
13
    connection = sqlite3.connect(DATABASE_PATH, check_same_thread=False)
14
    connection.row_factory = sqlite3.Row
15
    return connection
16

17
def init_db():
18
    conn = get_db_connection()
19
    conn.execute("""
20
        CREATE TABLE IF NOT EXISTS likes (
21
            id INTEGER PRIMARY KEY AUTOINCREMENT,
22
            page_name TEXT NOT NULL,
23
            guest_data TEXT NOT NULL,
24
            ts INTEGER NOT NULL,
25
            ip TEXT
26
        )
27
    """)
28
    conn.commit()
29
    conn.close()
30

31
init_db()
32

33
def get_db():
34
    connection = get_db_connection()
35
    db = connection.cursor()
36
    try:
37
        yield db
38
    finally:
39
        db.close()
40
        connection.close()
41

42
class Like(BaseModel):
43
    page_name: str
44
    guest_data: str
45
    ts: str | None = None
46
    ip: str | None = None
47

48
limiter = Limiter(key_func=get_remote_address)
49
app = FastAPI(docs_url=None, redoc_url=None)
50
app.state.limiter = limiter
51
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
52

53
app.add_middleware(
54
    CORSMiddleware,
55
    allow_origins=["https://blog.dotuoodo.top","http://localhost:1313"],
56
    allow_credentials=True,
57
    allow_methods=["*"],
58
    allow_headers=["*"],
59
)
60

61
@app.head("/")
62
@app.get("/")
63
def root():
64
    return "ok"
65

66
@app.get("/like/count")
67
def get_likes_count(
68
    page_name: str,
69
    db: sqlite3.Cursor = Depends(get_db)
70
):
71
    db.execute("SELECT COUNT(*) FROM likes WHERE page_name = ?", (page_name,))
72
    result = db.fetchone()
73
    return {"page": page_name, "likes": result[0] if result else 0}
74

75
@app.get("/like/pages")
76
def get_likes_pages(
77
    page_names: str,
78
    like_id: str | None = None,
79
    db: sqlite3.Cursor = Depends(get_db)
80
):
81
    page_list = page_names.split(',')
82
    placeholders = ', '.join(['?'] * len(page_list))
83
    query = f"""
84
        SELECT page_name, COUNT(*) AS likes
85
        FROM likes
86
        WHERE page_name IN ({placeholders})
87
        GROUP BY page_name
88
    """
89
    db.execute(query, tuple(page_list))
90
    results = db.fetchall()
91
    result_dict = {row[0]: row[1] for row in results}
92
    response_data = {name: result_dict.get(name, 0) for name in page_list}
93
    if like_id is not None:
94
        liked_query = f"""
95
            SELECT page_name
96
            FROM likes
97
            WHERE guest_data = ?
98
              AND page_name IN ({placeholders})
99
        """
100
        db.execute(liked_query, (like_id, *page_list))
101
        liked_pages = [row[0] for row in db.fetchall()]
102
        return {"data": response_data, "liked": liked_pages}
103
    return {"data": response_data}
104

105
@app.post("/like/toggle")
106
@limiter.limit("20/minute")
107
def toggle_like(
108
    request: Request,
109
    like: Like,
110
    db: sqlite3.Cursor = Depends(get_db)
111
):
112
    try:
113
        client_host = request.headers.get("CF-Connecting-IP")
114
        db.execute(
115
            "SELECT id FROM likes WHERE page_name = ? AND guest_data = ?",
116
            (like.page_name, like.guest_data)
117
        )
118
        existing = db.fetchone()
119
        if existing:
120
            db.execute("DELETE FROM likes WHERE id = ?", (existing[0],))
121
            action = "unliked"
122
        else:
123
            db.execute(
124
                "INSERT INTO likes (page_name, guest_data, ts, ip) VALUES (?, ?, ?, ?)",
125
                (like.page_name, like.guest_data, int(time.time()), client_host)
126
            )
127
            action = "liked"
128
        db.connection.commit()
129
        db.execute("SELECT COUNT(*) FROM likes WHERE page_name = ?", (like.page_name,))
130
        result = db.fetchone()
131
        return {"status": "success", "action": action, "like": result[0]}
132
    except Exception as e:
133
        db.connection.rollback()
134
        return {"status": "error", "detail": str(e)}