Something went wrong. Try again.
finger for the ATProto ecosystem atplan.io
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169import osimport httpxfrom datetime import datetime, timezonefrom dotenv import load_dotenvfrom flask import Flask, render_template, request, redirect, url_for, send_from_directory
load_dotenv()
PDS_HOST = os.getenv("PDS_HOST")HANDLE = os.getenv("HANDLE")PUBLICATION_URL = "https://atplan.io"
app = Flask(__name__)
def format_timestamp(ts): try: ts = ts.replace("Z", "+00:00") dt = datetime.fromisoformat(ts) dt = dt.astimezone(timezone.utc) return dt.strftime("%a %b %d %H:%M:%S UTC %Y") except Exception: return ts
def resolve_handle(handle): if handle.startswith("did:"): return handle response = httpx.get( f"{PDS_HOST}/xrpc/com.atproto.identity.resolveHandle", params={"handle": handle} ) if response.status_code != 200: raise Exception(f"Could not resolve handle: {handle}") return response.json().get("did")
def get_pds_for_did(did): try: if did.startswith("did:plc:"): response = httpx.get(f"https://plc.directory/{did}") if response.status_code == 200: doc = response.json() for service in doc.get("service", []): if service.get("id") == "#atproto_pds": return service.get("serviceEndpoint") except Exception: pass return PDS_HOST
def fetch_plan(did): pds = get_pds_for_did(did) response = httpx.get( f"{pds}/xrpc/com.atproto.repo.getRecord", params={ "repo": did, "collection": "io.atplan.plan", "rkey": "self" } ) if response.status_code == 200: return response.json().get("value", {}) return None
def fetch_record(did, collection, rkey): pds = get_pds_for_did(did) response = httpx.get( f"{pds}/xrpc/com.atproto.repo.getRecord", params={"repo": did, "collection": collection, "rkey": rkey} ) if response.status_code == 200: return response.json().get("value", {}) return None
def list_posts(did): pds = get_pds_for_did(did) response = httpx.get( f"{pds}/xrpc/com.atproto.repo.listRecords", params={"repo": did, "collection": "site.standard.document", "limit": 50} ) if response.status_code != 200: return [] posts = [] for record in response.json().get("records", []): rkey = record["uri"].split("/")[-1] value = record.get("value", {}) if rkey == "self" or value.get("site") != PUBLICATION_URL: continue posts.append({ "rkey": rkey, "title": value.get("title") or "(untitled)", "publishedAt": format_timestamp(value.get("publishedAt", "")) }) posts.sort(key=lambda p: p["rkey"], reverse=True) return posts
def get_plan_for_handle(handle): try: did = resolve_handle(handle) plan = fetch_plan(did) if plan and plan.get("updatedAt"): plan["updatedAt"] = format_timestamp(plan["updatedAt"]) if not plan: return None, handle, f"No .plan found for {handle}", did return plan, handle, None, did except Exception as e: return None, handle, str(e), None
@app.route("/")@app.route("/", methods=["GET", "POST"])def index(): if request.method == "POST": handle = request.form.get("handle", "").strip().lstrip("@") if handle: return redirect(url_for("finger", handle=handle)) owner_plan, owner_handle, owner_error, _ = get_plan_for_handle(HANDLE) return render_template("landing.html", owner_plan=owner_plan, owner_handle=owner_handle, owner_error=owner_error )
@app.route("/finger/<path:handle>")def finger(handle): handle = handle.lstrip("@") plan, handle, error, did = get_plan_for_handle(handle) posts = list_posts(did) if did else [] return render_template("finger.html", plan=plan, handle=handle, error=error, did=did, posts=posts)
@app.route("/finger/<path:handle>/post/<rkey>")def finger_post(handle, rkey): handle = handle.lstrip("@") try: did = resolve_handle(handle) except Exception as e: return render_template("post.html", post=None, handle=handle, error=str(e), did=None, rkey=rkey) post = fetch_record(did, "site.standard.document", rkey) if post and post.get("site") != PUBLICATION_URL: post = None if post and post.get("publishedAt"): post["publishedAt"] = format_timestamp(post["publishedAt"]) error = None if post else f"No post found for {handle}/{rkey}" return render_template("post.html", post=post, handle=handle, error=error, did=did, rkey=rkey)
@app.route("/edit")def edit(): return render_template("edit.html")
@app.route("/client-metadata.json")def client_metadata(): return send_from_directory("static", "client-metadata.json", mimetype="application/json")
@app.route("/lexicon/io.atplan.plan")def lexicon(): return send_from_directory("lexicon", "io.atplan.plan.json", mimetype="application/json")
@app.route("/.well-known/site.standard.publication")def standard_site_publication(): did = resolve_handle(HANDLE) return f"at://{did}/site.standard.publication/self", 200, {"Content-Type": "text/plain"}
@app.route("/embed/<path:handle>")def embed(handle): handle = handle.lstrip("@") plan, handle, error, did = get_plan_for_handle(handle) return render_template("embed.html", plan=plan, handle=handle, error=error, did=did)
if __name__ == "__main__": app.run(debug=True, port=5000)