import logging from typing import Annotated from sqlalchemy import or_ from fastapi import APIRouter, Request, Form from devplacepy.models import ProjectForm, ProjectEditForm, ProjectFlagForm, ForkForm from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from devplacepy.database import ( get_table, get_users_by_uids, get_site_stats, get_user_votes, paginate, resolve_by_slug, get_fork_parent, count_forks, ) from devplacepy.services.jobs import queue from devplacepy.content import ( load_detail, delete_content_item, create_content_item, edit_content_item, detail_context, canonical_redirect, first_image_url, is_owner, can_view_project, ) from devplacepy.utils import ( get_current_user, require_user, not_found, is_admin, XP_PROJECT, ) from devplacepy.seo import ( base_seo_context, site_url, website_schema, software_application_schema, list_page_seo, next_page_url, ) from devplacepy.responses import respond, action_result from devplacepy.schemas import ProjectsOut, ProjectDetailOut from devplacepy.services.audit import record as audit logger = logging.getLogger(__name__) router = APIRouter() def get_projects_list( tab: str = "recent", search: str = "", user_uid: str = None, project_type: str = None, before: str = None, viewer: dict = None, ): projects = get_table("projects") filters = {} if user_uid: filters["user_uid"] = user_uid if project_type: filters["project_type"] = project_type if tab == "released": filters["status"] = "Released" clauses = [] if projects.exists: columns = projects.table.columns if search: like = f"%{search}%" clauses.append( or_(columns.title.ilike(like), columns.description.ilike(like)) ) if not is_admin(viewer) and "is_private" in columns: visible = or_(columns.is_private.is_(None), columns.is_private == 0) if viewer: visible = or_(visible, columns.user_uid == viewer["uid"]) clauses.append(visible) order = ["-stars", "-created_at"] if tab == "popular" else ["-created_at"] total = projects.count(*clauses, **filters) page, next_cursor = paginate( projects, *clauses, before=before, order=order, **filters ) if page: users_map = get_users_by_uids([p["user_uid"] for p in page]) my_votes = ( get_user_votes(viewer["uid"], [p["uid"] for p in page]) if viewer else {} ) for p in page: author = users_map.get(p["user_uid"]) p["author_name"] = author["username"] if author else "Unknown" p["my_vote"] = my_votes.get(p["uid"], 0) return page, next_cursor, total @router.get("", response_class=HTMLResponse) async def projects_page( request: Request, tab: str = "recent", search: str = "", user_uid: str = None, project_type: str = None, before: str = None, ): user = get_current_user(request) projects, next_cursor, total_count = get_projects_list( tab, search, user_uid, project_type, before, viewer=user ) total_members = get_site_stats()["total_members"] seo_ctx = list_page_seo( request, title="Projects", description=f"Explore {total_count} developer projects on DevPlace. Games, software, mobile apps and more.", breadcrumbs=[ {"name": "Home", "url": "/feed"}, {"name": "Projects", "url": "/projects"}, ], next_url=next_page_url(request, next_cursor), ) return respond( request, "projects.html", { **seo_ctx, "request": request, "user": user, "projects": projects, "current_tab": tab, "search": search, "project_type": project_type, "total_count": total_count, "next_cursor": next_cursor, "total_members": total_members, }, model=ProjectsOut, ) @router.get("/{project_slug}", response_class=HTMLResponse) async def project_detail(request: Request, project_slug: str): user = get_current_user(request) detail = load_detail("projects", "project", project_slug, user) if not detail: raise not_found("Project not found") project = detail["item"] if not can_view_project(project, user): raise not_found("Project not found") redirect = canonical_redirect("projects", project, project_slug) if redirect: return redirect base = site_url(request) seo_ctx = base_seo_context( request, title=project.get("title", "Project"), description=project.get("description", "")[:160], og_image=first_image_url(project, detail["attachments"]), breadcrumbs=[ {"name": "Home", "url": "/feed"}, {"name": "Projects", "url": "/projects"}, { "name": project.get("title", "Project"), "url": f"/projects/{project['slug'] or project['uid']}", }, ], schemas=[website_schema(base), software_application_schema(project, base)], ) parent = get_fork_parent(project["uid"]) forked_from = ( { "uid": parent["uid"], "slug": parent.get("slug") or parent["uid"], "title": parent.get("title") or "project", } if parent else None ) return respond( request, "project_detail.html", detail_context( request, user, detail, "project", seo_ctx, { "platforms": project.get("platforms", "").split(",") if project.get("platforms") else [], "is_private": bool(project.get("is_private")), "read_only": bool(project.get("read_only")), "forked_from": forked_from, "fork_count": count_forks(project["uid"]), }, ), model=ProjectDetailOut, ) @router.post("/{project_slug}/zip") async def zip_project(request: Request, project_slug: str): project = resolve_by_slug(get_table("projects"), project_slug) if not project: raise not_found("Project not found") user = get_current_user(request) if not can_view_project(project, user): raise not_found("Project not found") owner_kind, owner_id = ( ("user", user["uid"]) if user else ("guest", request.client.host or "anon") ) uid = queue.enqueue( "zip", {"source": {"type": "project_tree", "project_uid": project["uid"], "path": ""}}, owner_kind, owner_id, project.get("title") or "project", ) audit.record( request, "project.zip.request", user=user, actor_kind="user" if user else "guest", target_type="project", target_uid=project["uid"], target_label=project.get("title"), summary=f"requested a zip of project {project.get('title')}", links=[audit.target("project", project["uid"], project.get("title")), audit.job(uid)], ) return JSONResponse({"uid": uid, "status_url": f"/zips/{uid}"}) @router.post("/{project_slug}/fork") async def fork_project( request: Request, project_slug: str, data: Annotated[ForkForm, Form()] ): user = require_user(request) source = resolve_by_slug(get_table("projects"), project_slug) if not source: raise not_found("Project not found") if not can_view_project(source, user): raise not_found("Project not found") title = data.title.strip() uid = queue.enqueue( "fork", { "source_project_uid": source["uid"], "title": title, "forked_by_uid": user["uid"], }, "user", user["uid"], title, ) audit.record( request, "project.fork.request", user=user, target_type="project", target_uid=source["uid"], target_label=source.get("title"), metadata={"new_title": title}, summary=f"{user['username']} requested a fork of project {source.get('title')}", links=[audit.source("project", source["uid"], source.get("title")), audit.job(uid)], ) return JSONResponse({"uid": uid, "status_url": f"/forks/{uid}"}) @router.post("/delete/{project_slug}") async def delete_project(request: Request, project_slug: str): user = require_user(request) return delete_content_item( request, "projects", "project", user, project_slug, "/projects" ) @router.post("/create") async def create_project(request: Request, data: Annotated[ProjectForm, Form()]): user = require_user(request) title = data.title.strip() description = data.description.strip() uid, project_slug = create_content_item( "projects", "project", user, { "title": title, "description": description, "release_date": data.release_date or None, "demo_date": data.demo_date or None, "project_type": data.project_type, "platforms": data.platforms.strip(), "status": data.status, "is_private": 1 if data.is_private else 0, "read_only": 0, }, title, XP_PROJECT, "First Project", description, data.attachment_uids, request, ) url = f"/projects/{project_slug}" return action_result( request, url, data={"uid": uid, "slug": project_slug, "url": url} ) @router.post("/edit/{project_slug}") async def edit_project( request: Request, project_slug: str, data: Annotated[ProjectEditForm, Form()] ): user = require_user(request) return edit_content_item( request, "projects", user, project_slug, { "title": data.title.strip(), "description": data.description.strip(), "release_date": data.release_date or None, "demo_date": data.demo_date or None, "project_type": data.project_type, "platforms": data.platforms.strip(), "status": data.status, }, "/projects", target_type="project", ) _FLAG_EVENTS = { ("is_private", True): "project.visibility.private", ("is_private", False): "project.visibility.public", ("read_only", True): "project.readonly.enable", ("read_only", False): "project.readonly.disable", } def _set_project_flag(request: Request, project_slug: str, field: str, value: bool): user = require_user(request) projects = get_table("projects") project = resolve_by_slug(projects, project_slug) event_key = _FLAG_EVENTS[(field, bool(value))] if not is_owner(project, user): from devplacepy.responses import wants_json, json_error audit.record( request, event_key, user=user, result="denied", target_type="project", target_uid=project["uid"] if project else project_slug, target_label=project.get("title") if project else project_slug, summary=f"{user['username']} denied changing {field} on project {project_slug}", ) if wants_json(request): return json_error(403, "Not allowed") return RedirectResponse(url=f"/projects/{project_slug}", status_code=302) old_value = 1 if project.get(field) else 0 projects.update({"uid": project["uid"], field: 1 if value else 0}, ["uid"]) logger.info( "project %s %s set to %s by %s", project["uid"], field, bool(value), user["username"], ) audit.record( request, event_key, user=user, target_type="project", target_uid=project["uid"], target_label=project.get("title"), old_value=old_value, new_value=1 if value else 0, summary=f"{user['username']} set {field} of project {project.get('title')} to {bool(value)}", links=[audit.target("project", project["uid"], project.get("title"))], ) url = f"/projects/{project['slug'] or project['uid']}" return action_result( request, url, data={"uid": project["uid"], field: bool(value), "url": url} ) @router.post("/{project_slug}/private") async def set_project_private( request: Request, project_slug: str, data: Annotated[ProjectFlagForm, Form()] ): return _set_project_flag(request, project_slug, "is_private", data.value) @router.post("/{project_slug}/readonly") async def set_project_readonly( request: Request, project_slug: str, data: Annotated[ProjectFlagForm, Form()] ): return _set_project_flag(request, project_slug, "read_only", data.value)