import logging
from typing import Annotated
from sqlalchemy import or_
from fastapi import APIRouter, Request, Form
from devplacepy.models import ProjectForm, ProjectFlagForm
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
from devplacepy.services.jobs import queue
from devplacepy.content import load_detail, delete_content_item, create_content_item, detail_context, canonical_redirect, first_image_url, is_owner, can_view_project
from devplacepy.templating import templates
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
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 == 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)],
)
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")),
}), 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",
)
return JSONResponse({"uid": uid, "status_url": f"/zips/{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)
url = f"/projects/{project_slug}"
return action_result(request, url, data={"uid": uid, "slug": project_slug, "url": url})
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)
if not is_owner(project, user):
from devplacepy.responses import wants_json, json_error
if wants_json(request):
return json_error(403, "Not allowed")
return RedirectResponse(url=f"/projects/{project_slug}", status_code=302)
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"])
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)
View Git Blame Copy Permalink