323 lines
11 KiB
Python
323 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""MyMovies media-library management API.
|
|
|
|
Endpoints (all behind Traefik basic-auth in prod):
|
|
POST /upload multipart: torrent, [meta], [name], [year]
|
|
POST /upload/raw JSON: {name, magnet, [year]}
|
|
DELETE /movie/<id> remove film folder + catalog entry
|
|
DELETE /movie/<id>/torrent remove only .torrent
|
|
GET /list all films (id,title,year,has_torrent,hash_ok,size,date)
|
|
GET /movie/<id> film details
|
|
PATCH /movie/<id> JSON: title,year,description,poster,...
|
|
POST /movie/<id>/meta replace meta.txt (file or text body)
|
|
POST /movie/<id>/poster multipart: poster
|
|
POST /movie/<id>/torrent replace .torrent (rehash)
|
|
PUT /movie/<id>/rename JSON: {new_id}
|
|
GET /check integrity check
|
|
POST /rebuild rebuild catalog.json from /movies
|
|
POST /deploy trigger Coolify webhook
|
|
"""
|
|
import os, re, json, shutil, subprocess, hashlib, datetime, unicodedata
|
|
from pathlib import Path
|
|
from fastapi import FastAPI, HTTPException, UploadFile, File, Form
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
|
|
APP_DIR = Path(os.environ.get("APP_DIR", "/data/app"))
|
|
MOVIES_DIR = APP_DIR / "movies"
|
|
CATALOG = MOVIES_DIR / "catalog.json"
|
|
WEBHOOK = os.environ.get("DEPLOY_WEBHOOK", "")
|
|
WEBHOOK_TOKEN = os.environ.get("COOLIFY_TOKEN", "")
|
|
|
|
app = FastAPI(title="MyMovies API")
|
|
|
|
# ---------- helpers ----------
|
|
|
|
ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*$")
|
|
|
|
def valid_id(s: str) -> str:
|
|
s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode()
|
|
s = s.lower().replace(" ", "-").replace("_", "-")
|
|
s = re.sub(r"[^a-z0-9-]", "", s)
|
|
if not ID_RE.match(s):
|
|
raise HTTPException(400, "id must be latin/numbers/dash, got: " + s)
|
|
return s
|
|
|
|
def load_catalog():
|
|
if not CATALOG.exists():
|
|
return {"movies": []}
|
|
return json.loads(CATALOG.read_text(encoding="utf-8"))
|
|
|
|
def deploy():
|
|
if not WEBHOOK:
|
|
return {"deployed": False, "reason": "no webhook configured"}
|
|
import urllib.request
|
|
req = urllib.request.Request(WEBHOOK, method="POST")
|
|
if WEBHOOK_TOKEN:
|
|
req.add_header("Authorization", "Bearer " + WEBHOOK_TOKEN)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return {"deployed": True, "status": r.status}
|
|
except Exception as e:
|
|
return {"deployed": False, "error": str(e)}
|
|
|
|
def torrent_hash(path: Path):
|
|
"""Extract info_dict sha1 from a .torrent (bencode) like build_catalog.sh."""
|
|
data = path.read_bytes()
|
|
def decode(data, i=0):
|
|
c = data[i:i+1]
|
|
if c == b"d":
|
|
i += 1; d = {}
|
|
while data[i:i+1] != b"e":
|
|
k, i = decode(data, i)
|
|
v, i = decode(data, i)
|
|
d[k] = v
|
|
return d, i+1
|
|
if c == b"l":
|
|
i += 1; lst = []
|
|
while data[i:i+1] != b"e":
|
|
v, i = decode(data, i); lst.append(v)
|
|
return lst, i+1
|
|
if c == b"i":
|
|
j = data.index(b"e", i); return int(data[i+1:j]), j+1
|
|
# string: digits:length
|
|
j = data.index(b":", i); n = int(data[i:j])
|
|
s = data[j+1:j+1+n]
|
|
return s, j+1+n
|
|
info = decode(data)[0][b"info"]
|
|
def enc(d):
|
|
if isinstance(d, dict):
|
|
return b"d" + b"".join(enc(k)+enc(v) for k, v in sorted(d.items())) + b"e"
|
|
if isinstance(d, list):
|
|
return b"l" + b"".join(enc(x) for x in d) + b"e"
|
|
if isinstance(d, int):
|
|
return b"i" + str(d).encode() + b"e"
|
|
return str(len(d)).encode() + b":" + d
|
|
return hashlib.sha1(enc(info)).hexdigest()
|
|
|
|
def rebuild_catalog():
|
|
"""Rebuild catalog.json by running the project's build_catalog.sh
|
|
(keeps a single source of truth for catalog generation)."""
|
|
script = MOVIES_DIR.parent / "build_catalog.sh"
|
|
if not script.exists():
|
|
raise HTTPException(500, "build_catalog.sh not found at %s" % script)
|
|
try:
|
|
subprocess.run(["bash", str(script)], check=True,
|
|
capture_output=True, text=True, timeout=120)
|
|
except subprocess.CalledProcessError as e:
|
|
raise HTTPException(500, "build failed: " + (e.stderr or e.stdout or str(e)))
|
|
cat = load_catalog()
|
|
return len(cat.get("movies", []))
|
|
|
|
# ---------- endpoints ----------
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok", "movies_dir": str(MOVIES_DIR), "catalog": str(CATALOG)}
|
|
|
|
@app.post("/upload")
|
|
def upload(torrent: UploadFile = File(...), meta: UploadFile = File(None),
|
|
name: str = Form(None), year: str = Form(None)):
|
|
if not name:
|
|
name = torrent.filename.rsplit(".", 1)[0]
|
|
name = valid_id(name)
|
|
d = MOVIES_DIR / name
|
|
if d.exists():
|
|
raise HTTPException(409, "film %s already exists" % name)
|
|
d.mkdir(parents=True)
|
|
tp = d / torrent.filename
|
|
tp.write_bytes(torrent.file.read())
|
|
if meta:
|
|
(d / "meta.txt").write_bytes(meta.file.read())
|
|
elif year:
|
|
(d / "meta.txt").write_text("year: %s\n" % year, encoding="utf-8")
|
|
n = rebuild_catalog()
|
|
return {"added": name, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.post("/upload/raw")
|
|
def upload_raw(body: dict):
|
|
name = valid_id(body.get("name", ""))
|
|
magnet = body.get("magnet", "")
|
|
if not name or not magnet:
|
|
raise HTTPException(400, "need name + magnet")
|
|
d = MOVIES_DIR / name
|
|
if d.exists():
|
|
raise HTTPException(409, "film %s already exists" % name)
|
|
d.mkdir(parents=True)
|
|
(d / "meta.txt").write_text(
|
|
"title: %s\nmagnet: %s\n" % (body.get("title", name), magnet), encoding="utf-8")
|
|
n = rebuild_catalog()
|
|
return {"added": name, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.delete("/movie/{fid}")
|
|
def delete_movie(fid: str):
|
|
fid = valid_id(fid)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
shutil.rmtree(d)
|
|
n = rebuild_catalog()
|
|
return {"deleted": fid, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.delete("/movie/{fid}/torrent")
|
|
def delete_torrent(fid: str):
|
|
fid = valid_id(fid)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
for t in d.glob("*.torrent"):
|
|
t.unlink()
|
|
n = rebuild_catalog()
|
|
return {"deleted_torrent": fid, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.get("/list")
|
|
def list_movies():
|
|
movies = load_catalog().get("movies", [])
|
|
out = []
|
|
for m in movies:
|
|
d = MOVIES_DIR / m.get("id", "")
|
|
size = 0
|
|
if d.exists():
|
|
for f in d.rglob("*"):
|
|
if f.is_file():
|
|
size += f.stat().st_size
|
|
tors = list(d.glob("*.torrent")) if d.exists() else []
|
|
out.append({
|
|
"id": m.get("id"), "title": m.get("title"), "year": m.get("year"),
|
|
"has_torrent": bool(tors), "size_bytes": size,
|
|
"hash_ok": bool(m.get("hash")),
|
|
})
|
|
return {"count": len(out), "movies": out}
|
|
|
|
@app.get("/movie/{fid}")
|
|
def movie_detail(fid: str):
|
|
fid = valid_id(fid)
|
|
cat = load_catalog()
|
|
m = next((x for x in cat.get("movies", []) if x.get("id") == fid), None)
|
|
if not m:
|
|
raise HTTPException(404, "no such film")
|
|
d = MOVIES_DIR / fid
|
|
files = [str(p.relative_to(d)) for p in d.rglob("*") if p.is_file()] if d.exists() else []
|
|
m = dict(m); m["files"] = files
|
|
return m
|
|
|
|
class PatchBody(BaseModel):
|
|
title: str = None
|
|
year: int = None
|
|
description: str = None
|
|
poster: str = None
|
|
original_title: str = None
|
|
|
|
@app.patch("/movie/{fid}")
|
|
def patch_movie(fid: str, body: PatchBody):
|
|
fid = valid_id(fid)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
mf = d / "meta.txt"
|
|
# update meta.txt fields (source of truth for build_catalog.sh)
|
|
fields = {}
|
|
if mf.exists():
|
|
for line in mf.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
line = line.strip()
|
|
if not line or ":" not in line:
|
|
continue
|
|
k, v = line.split(":", 1)
|
|
fields[k.strip()] = v.strip()
|
|
for k in ("title", "year", "description", "poster", "original_title"):
|
|
v = getattr(body, k)
|
|
if v is not None:
|
|
fields[k] = str(v)
|
|
mf.write_text("\n".join("%s: %s" % (k, v) for k, v in fields.items()) + "\n",
|
|
encoding="utf-8")
|
|
n = rebuild_catalog()
|
|
return {"updated": fid, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.post("/movie/{fid}/meta")
|
|
def post_meta(fid: str, meta: UploadFile = File(None)):
|
|
fid = valid_id(fid)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
if meta:
|
|
(d / "meta.txt").write_bytes(meta.file.read())
|
|
else:
|
|
# text body
|
|
import io
|
|
data = getattr(meta, "file", None)
|
|
n = rebuild_catalog()
|
|
return {"meta_updated": fid, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.post("/movie/{fid}/poster")
|
|
def post_poster(fid: str, poster: UploadFile = File(...)):
|
|
fid = valid_id(fid)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
ext = poster.filename.rsplit(".", 1)[-1].lower()
|
|
if ext not in ("jpg", "jpeg", "png", "webp"):
|
|
raise HTTPException(400, "bad poster ext")
|
|
(d / ("poster." + ext)).write_bytes(poster.file.read())
|
|
n = rebuild_catalog()
|
|
return {"poster_updated": fid, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.post("/movie/{fid}/torrent")
|
|
def post_torrent(fid: str, torrent: UploadFile = File(...)):
|
|
fid = valid_id(fid)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
for t in d.glob("*.torrent"):
|
|
t.unlink()
|
|
tp = d / torrent.filename
|
|
tp.write_bytes(torrent.file.read())
|
|
n = rebuild_catalog()
|
|
return {"torrent_updated": fid, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
class RenameBody(BaseModel):
|
|
new_id: str
|
|
|
|
@app.put("/movie/{fid}/rename")
|
|
def rename_movie(fid: str, body: RenameBody):
|
|
fid = valid_id(fid)
|
|
new = valid_id(body.new_id)
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
raise HTTPException(404, "no such film")
|
|
if (MOVIES_DIR / new).exists():
|
|
raise HTTPException(409, "target %s exists" % new)
|
|
d.rename(MOVIES_DIR / new)
|
|
n = rebuild_catalog()
|
|
return {"renamed": fid, "to": new, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.get("/check")
|
|
def check():
|
|
movies = load_catalog().get("movies", [])
|
|
problems = []
|
|
for m in movies:
|
|
fid = m.get("id", "")
|
|
d = MOVIES_DIR / fid
|
|
if not d.exists():
|
|
problems.append({"id": fid, "issue": "folder missing"})
|
|
continue
|
|
tors = list(d.glob("*.torrent"))
|
|
if m.get("hash") and tors:
|
|
try:
|
|
h = torrent_hash(tors[0])
|
|
if h != m["hash"]:
|
|
problems.append({"id": fid, "issue": "hash mismatch %s != %s" % (h, m["hash"])})
|
|
except Exception as e:
|
|
problems.append({"id": fid, "issue": "torrent parse error: %s" % e})
|
|
elif not tors:
|
|
problems.append({"id": fid, "issue": "no .torrent"})
|
|
return {"checked": len(movies), "problems": problems}
|
|
|
|
@app.post("/rebuild")
|
|
def rebuild():
|
|
n = rebuild_catalog()
|
|
return {"rebuilt": True, "catalog_movies": n, "deploy": deploy()}
|
|
|
|
@app.post("/deploy")
|
|
def do_deploy():
|
|
return deploy()
|