init linker
This commit is contained in:
commit
bf3419a7f0
11 changed files with 4268 additions and 0 deletions
152
linker.py
Executable file
152
linker.py
Executable file
|
|
@ -0,0 +1,152 @@
|
|||
#!/usr/bin/env -S uv run --quiet --script
|
||||
# /// script
|
||||
# requires-python = ">=3.12"
|
||||
# dependencies = [
|
||||
# "fastapi",
|
||||
# "uvicorn[standard]",
|
||||
# "jinja2",
|
||||
# "httpx",
|
||||
# "beautifulsoup4",
|
||||
# "python-slugify",
|
||||
# ]
|
||||
# ///
|
||||
|
||||
from pathlib import Path
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import HTMLResponse
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from urllib.parse import urlencode
|
||||
import httpx
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
app = FastAPI()
|
||||
templates = Jinja2Templates(directory="templates")
|
||||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||||
|
||||
VERSION = Path("version").read_text()
|
||||
|
||||
|
||||
def get_meta_tag(soup: BeautifulSoup, names: list[str], attrs: list[str]) -> str | None:
|
||||
for name in names:
|
||||
for attr in attrs:
|
||||
tag = soup.find("meta", attrs={attr: name})
|
||||
if tag and tag.get("content"):
|
||||
return tag["content"]
|
||||
return None
|
||||
|
||||
|
||||
def extract_metadata(html: str) -> dict:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
|
||||
title = (
|
||||
get_meta_tag(soup, ["og:title"], ["property"]) or soup.title.string
|
||||
if soup.title
|
||||
else None
|
||||
)
|
||||
image = (
|
||||
get_meta_tag(soup, ["og:image", "twitter:image"], ["property", "name"]) or None
|
||||
)
|
||||
author = get_meta_tag(
|
||||
soup, ["author", "article:author", "twitter:creator"], ["name", "property"]
|
||||
)
|
||||
|
||||
description = get_meta_tag(
|
||||
soup, ["description", "og:description"], ["name", "property"]
|
||||
)
|
||||
|
||||
return {
|
||||
"title": title,
|
||||
"image": image,
|
||||
"author": author,
|
||||
"description": description,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
async def index(request: Request):
|
||||
return templates.TemplateResponse(
|
||||
"index.html", {"request": request, "version": VERSION}
|
||||
)
|
||||
|
||||
|
||||
@app.get("/link", response_class=HTMLResponse)
|
||||
async def link_preview(request: Request, url: str):
|
||||
async with httpx.AsyncClient(follow_redirects=True) as client:
|
||||
try:
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
except httpx.HTTPError:
|
||||
return templates.TemplateResponse(
|
||||
"error.html",
|
||||
{
|
||||
"request": request,
|
||||
"url": url,
|
||||
"error": f"An error occurred while fetching the page: {url}",
|
||||
"version": VERSION,
|
||||
},
|
||||
status_code=400,
|
||||
)
|
||||
|
||||
meta = extract_metadata(response.text)
|
||||
|
||||
preview_image = (
|
||||
meta["image"]
|
||||
or f"http://shots.wayl.one/shot/?{urlencode({'url': url, 'height': 450, 'width': 800, 'scaled_width': 800, 'scaled_height': 450, 'selectors': ''})}"
|
||||
)
|
||||
|
||||
title_author = meta["title"] or url
|
||||
if meta.get("author"):
|
||||
title_author += f" by {meta['author']}"
|
||||
|
||||
sub_text = f" {meta.get('title')}"
|
||||
if meta.get("author"):
|
||||
sub_text += f" by {meta.get('author')}"
|
||||
if meta.get("description"):
|
||||
sub_text += f"- {meta.get('description')}"
|
||||
|
||||
markdown_link = f"""
|
||||
!!! seealso ""
|
||||
[]({url})
|
||||
{sub_text.strip()}
|
||||
"""
|
||||
|
||||
return templates.TemplateResponse(
|
||||
"link.html",
|
||||
{
|
||||
"request": request,
|
||||
"url": url,
|
||||
"markdown_link": markdown_link,
|
||||
"preview_image": preview_image,
|
||||
"title_author": title_author,
|
||||
"version": VERSION,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
def health_check():
|
||||
# Minimal check to confirm the app is running.
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@app.get("/ready")
|
||||
def readiness_check():
|
||||
# Add additional checks (e.g. database connectivity) as needed.
|
||||
# For example:
|
||||
# if not database_is_connected():
|
||||
# raise HTTPException(status_code=503, detail="Service Unavailable")
|
||||
return {"status": "ready"}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
# get port from cli
|
||||
import sys
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
port = int(sys.argv[1])
|
||||
uvicorn.run(app, host="0.0.0.0", port=port)
|
||||
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
Loading…
Add table
Add a link
Reference in a new issue