190 lines
3.8 KiB
Python
190 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
# Standard
|
|
import random
|
|
import string
|
|
import datetime
|
|
from typing import Any
|
|
|
|
# Modules
|
|
import db as DB
|
|
import config as Config
|
|
|
|
|
|
def get_value(curl: str, what: str) -> Any:
|
|
dbase = DB.get_db()
|
|
cursor = dbase.cursor()
|
|
db_string = f"SELECT {what} FROM curls WHERE curl = ?"
|
|
cursor.execute(db_string, (curl,))
|
|
return cursor.fetchone()
|
|
|
|
|
|
def add_curl(curl: str, key: str) -> None:
|
|
dbase = DB.get_db()
|
|
cursor = dbase.cursor()
|
|
|
|
db_string = """
|
|
INSERT INTO curls (created, updated, curl, key, status)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
"""
|
|
|
|
now = date_now()
|
|
cursor.execute(db_string, (now, now, curl, key, ""))
|
|
dbase.commit()
|
|
|
|
|
|
def make_key(curl: str) -> str:
|
|
characters = string.ascii_letters + string.digits
|
|
chars = "".join(random.choice(characters) for i in range(Config.key_length))
|
|
start = curl[:3]
|
|
rest = len(start) + 1
|
|
return f"{start}_{chars[rest:]}"
|
|
|
|
|
|
def check_key(curl: str, key: str) -> bool:
|
|
if not key:
|
|
return False
|
|
|
|
if len(key) > Config.key_length:
|
|
return False
|
|
|
|
result = get_value(curl, "key")
|
|
return bool(result) and (result[0] == key)
|
|
|
|
|
|
def change_status(curl: str, status: str) -> None:
|
|
current = get_status(curl, False)
|
|
|
|
if current and (current == status):
|
|
return
|
|
|
|
dbase = DB.get_db()
|
|
cursor = dbase.cursor()
|
|
|
|
db_string = """
|
|
UPDATE curls
|
|
SET status = ?, updated = ?, changes = changes + 1
|
|
WHERE curl = ?
|
|
"""
|
|
|
|
now = date_now()
|
|
cursor.execute(db_string, (status, now, curl))
|
|
dbase.commit()
|
|
|
|
|
|
def curl_exists(curl: str) -> bool:
|
|
result = get_value(curl, "curl")
|
|
return bool(result)
|
|
|
|
|
|
def get_status(curl: str, fill: bool = True) -> str:
|
|
result = get_value(curl, "status")
|
|
|
|
if fill:
|
|
return fill_status(result)
|
|
|
|
if result:
|
|
return str(result[0])
|
|
|
|
return ""
|
|
|
|
|
|
def get_curl_list(curls: list[str]) -> list[dict[str, Any]]:
|
|
dbase = DB.get_db()
|
|
cursor = dbase.cursor()
|
|
|
|
db_string = """
|
|
SELECT created, curl, status, updated, changes
|
|
FROM curls
|
|
WHERE curl IN ({})
|
|
""".format(",".join("?" * len(curls)))
|
|
|
|
cursor.execute(db_string, curls)
|
|
results = cursor.fetchall()
|
|
items = []
|
|
|
|
for result in results:
|
|
if not result:
|
|
continue
|
|
|
|
created = str(result[0]) or ""
|
|
curl = result[1]
|
|
status = result[2]
|
|
updated = str(result[3]) or ""
|
|
changes = result[4] or 0
|
|
|
|
items.append(
|
|
{
|
|
"created": created,
|
|
"curl": curl,
|
|
"status": status,
|
|
"updated": updated,
|
|
"changes": changes,
|
|
}
|
|
)
|
|
|
|
return items
|
|
|
|
|
|
def fill_status(result: Any) -> str:
|
|
if not result:
|
|
return "Not claimed yet"
|
|
|
|
status = result[0]
|
|
|
|
if not status:
|
|
return "Not updated yet"
|
|
|
|
return str(status)
|
|
|
|
|
|
def curl_too_long() -> str:
|
|
return f"Error: Curl is too long (Max is {Config.curl_max_length} characters)"
|
|
|
|
|
|
def status_too_long() -> str:
|
|
return f"Error: Text is too long (Max is {Config.status_max_length} characters)"
|
|
|
|
|
|
def too_many_curls() -> str:
|
|
return f"Error: Too many curls (Max is {Config.max_curls})"
|
|
|
|
|
|
def check_curl(curl: str) -> bool:
|
|
if not curl:
|
|
return False
|
|
|
|
if len(curl) > Config.curl_max_length:
|
|
return False
|
|
|
|
if not curl.isalnum():
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def check_status(status: str) -> bool:
|
|
if not status:
|
|
return False
|
|
|
|
if len(status) > Config.status_max_length:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def date_now() -> datetime.datetime:
|
|
return datetime.datetime.now(datetime.timezone.utc)
|
|
|
|
|
|
def clean_curl(curl: str) -> str:
|
|
return str(curl).strip().lower()
|
|
|
|
|
|
def clean_key(key: str) -> str:
|
|
return str(key).strip()
|
|
|
|
|
|
def clean_status(status: str) -> str:
|
|
return str(status).strip()
|