"""Exact-job private photos and reviewed public gallery snapshots. No external I/O."""
import hashlib
import io
import json
import os
import re
import sqlite3
from pathlib import Path
from html import escape
from fastapi import HTTPException,Request
from fastapi.responses import FileResponse,HTMLResponse
from PIL import Image,ImageOps,UnidentifiedImageError
from . import photo_models as m
from .identity import assigned
from .store import uid,now,one,many
from .site_models import public_content
MAX_UPLOAD=2*1024*1024
MAX_JOB_PHOTOS=100
MAX_BUSINESS_PHOTOS=1000
MAX_PROJECTS=100
def fail(status,message):raise HTTPException(status,message)
def required(row):
if not row:fail(404,'Photo or project not found in this business.')
return row
def decode_image(raw):
if not raw or len(raw)>MAX_UPLOAD:fail(413,'Use a PNG, JPEG or WebP image no larger than 2 MB.')
try:
source=Image.open(io.BytesIO(raw))
if source.format not in ('PNG','JPEG','WEBP') or getattr(source,'n_frames',1)!=1 or source.width*source.height>12_000_000:fail(415,'Use a single PNG, JPEG or WebP image up to 12 megapixels.')
source.load();source=ImageOps.exif_transpose(source);source.thumbnail((1800,1800))
clean=Image.new('RGBA',source.size);clean.paste(source.convert('RGBA'))
output=io.BytesIO();clean.save(output,'PNG');body=output.getvalue()
return body,clean.width,clean.height
except (UnidentifiedImageError,OSError,ValueError,Image.DecompressionBombError,Image.DecompressionBombWarning):fail(415,'That file is not a supported image.')
def job_access(db,s,job_id):
job=required(one(db,'SELECT id,customer_id FROM jobs WHERE id=?',(job_id,)))
if s['role']!='owner' and (s['role']!='staff' or not assigned(db,s,job_id)):fail(404,'Assigned job not available.')
return job
def photo_access(db,s,job_id,photo_id,*,archived=False):
job=job_access(db,s,job_id)
row=required(one(db,'SELECT * FROM job_photos WHERE id=? AND job_id=? AND customer_id=?',(photo_id,job_id,job['customer_id'])))
if s['role']!='owner' and row['uploaded_by']=='public-enquiry' and not row['reviewed']:fail(404,'Photo not available until the owner reviews it.')
if row['archived'] and not (archived and s['role']=='owner'):fail(404,'Photo not available.')
return row
def photo_snapshot(db,photo_id,s,action):
photo=required(one(db,'SELECT * FROM job_photos WHERE id=?',(photo_id,)))
db.execute('INSERT INTO photo_history(photo_id,action,actor_id,membership_id,snapshot,created_at) VALUES(?,?,?,?,?,?)',(photo_id,action,s['subject'],s['membership_id'],json.dumps(photo),now()))
def project_snapshot(db,project_id,s,action,permission=False):
row=required(one(db,'SELECT * FROM gallery_projects WHERE id=?',(project_id,)))
db.execute('INSERT INTO gallery_history(project_id,action,actor_id,membership_id,version,snapshot,permission_confirmed,created_at) VALUES(?,?,?,?,?,?,?,?)',(project_id,action,s['subject'],s['membership_id'],row['version'],json.dumps(row),int(permission),now()))
def photo_projection(row,owner=False):
fields=('id','job_id','caption','description','category','phase','pair_id','position','reviewed','archived','version','created_at','updated_at')
result={k:row[k] for k in fields}
result['origin']='public_enquiry' if row['uploaded_by']=='public-enquiry' else 'workspace'
if owner:result.update(uploaded_by=row['uploaded_by'],membership_id=row['membership_id'],customer_share=json.loads(row['customer_share']) if row['customer_share'] else None)
return result
def active_project_references(db,photo_id):
result=[]
for row in many(db,'SELECT id,published FROM gallery_projects WHERE published IS NOT NULL AND archived=0'):
content=json.loads(row['published'])
if any(i['source_kind']=='job' and i['source_id']==photo_id for i in content['images']):result.append({'id':row['id'],'title':content['title']})
return result
def project_row(db,project_id,*,archived=False):
row=required(one(db,'SELECT * FROM gallery_projects WHERE id=?',(project_id,)))
if row['archived'] and not archived:fail(409,'Restore this archived project before changing it.')
return row
def project_sources(db,row,content,*,publishing=False,check_slug=True):
if check_slug:
for other in many(db,'SELECT id,draft_slug,published FROM gallery_projects WHERE id<>?',(row['id'],)):
if other['draft_slug']==content['slug'] or (other['published'] and json.loads(other['published'])['slug']==content['slug']):fail(409,'That project address is already reserved. Choose another.')
services={s['slug'] for s in json.loads(db.execute('SELECT published FROM site').fetchone()[0])['services'] if s.get('visible',True)}
if publishing and not set(content['service_slugs'])<=services:fail(422,'Publish the selected services first, or remove unavailable service associations.')
pairs={}
for image in content['images']:
if image['source_kind']=='job':
photo=required(one(db,'SELECT p.*,j.customer_id AS current_customer FROM job_photos p JOIN jobs j ON j.id=p.job_id WHERE p.id=? AND p.job_id=?',(image['source_id'],row['source_job_id'])))
if photo['customer_id']!=photo['current_customer']:fail(409,'A photo has an inconsistent customer link.')
if publishing and (photo['archived'] or not photo['reviewed']):fail(409,'Review each selected job photo and restore archived photos before publishing.')
else:
if row['source_job_id']:fail(422,'A job project must select images from that exact job.')
required(one(db,'SELECT id FROM assets WHERE id=?',(image['source_id'],)))
if publishing and not image['alt']:fail(422,'Give every public image a separate image description.')
if image['pair_id']:pairs.setdefault(image['pair_id'],set()).add(image['phase'])
if publishing:
if not content['images']:fail(422,'Choose at least one reviewed image before publishing.')
if any(phases!={'before','after'} for phases in pairs.values()):fail(422,'Each published comparison needs one Before and one After image.')
# Existing public image IDs keep their original source through draft edits/history.
prior=[json.loads(h['snapshot']) for h in many(db,'SELECT snapshot FROM gallery_history WHERE project_id=?',(row['id'],))]
prior.append(row)
links={}
for state in prior:
for key in ('draft','published'):
raw=state.get(key)
if not raw:continue
for i in json.loads(raw)['images']:links[i['id']]=(i['source_kind'],i['source_id'])
for image in content['images']:
if image['id'] in links and links[image['id']]!=(image['source_kind'],image['source_id']):fail(422,'Use a new image identity when choosing a different source photograph.')
def project_projection(content,services=None):
fields=('title','slug','description','category','meta_description','noindex','in_index','position')
result={k:content[k] for k in fields}
result['service_slugs']=[s for s in content['service_slugs'] if services is None or s in services]
result['images']=[{k:i[k] for k in ('id','caption','alt','phase','pair_id')} for i in content['images']]
return result
def public_projects(db,*,indexed=True):
site=public_content(json.loads(db.execute('SELECT published FROM site').fetchone()[0]));services={s['slug'] for s in site['services']}
projects=[]
for row in many(db,'SELECT published FROM gallery_projects WHERE published IS NOT NULL AND archived=0'):
content=json.loads(row['published'])
if not indexed or content['in_index']:projects.append(project_projection(content,services))
return sorted(projects,key=lambda p:(p['position'],p['title'].lower(),p['slug']))
def project_paths(db):return ['work/projects',*('work/projects/'+p['slug'] for p in public_projects(db,indexed=False))]
def published_project(db,slug):
for row in many(db,'SELECT * FROM gallery_projects WHERE published IS NOT NULL AND archived=0'):
content=json.loads(row['published'])
if content['slug']==slug:return row,content
fail(404,'This project is not published.')
def source_path(store,db,bid,image):
if image['source_kind']=='website':
required(one(db,'SELECT id FROM assets WHERE id=?',(image['source_id'],)))
return store.asset_path(bid,image['source_id'])
photo=required(one(db,'SELECT blob_id FROM job_photos WHERE id=?',(image['source_id'],)))
return store.photo_path(bid,photo['blob_id'])
def gallery_shell(static_dir,site,project,base,projects):
title=(project['title'] if project else 'Project gallery')+' · '+site['business_name']
description=(project['meta_description'] or project['description']) if project else 'Published projects from '+site['business_name']
canonical=base+('/'+project['slug'] if project else '')
tags=''+escape(title)+''
for key,value in [('og:title',title),('og:description',description),('og:url',canonical),('og:type','website'),('og:site_name',site['business_name'])]:tags+=''
if project and project['images']:
image_url=base.replace('/site/','/api/public/').replace('/work/projects','/projects')+'/'+project['slug']+'/images/'+project['images'][0]['id']
tags+=''
fallback=''
html=(Path(static_dir)/'index.html').read_text();html=re.sub(r'.*?','',html,flags=re.S);html=re.sub(r']*>','',html)
return html.replace('',tags+'').replace('
',fallback+'')
def mount_photos(app,store,session,once,business,portal_session,static_dir):
@app.get('/api/jobs/{job_id}/photos')
def photos(job_id:str,request:Request,archived:bool=False):
s=session(request)
with store.tenant(s['business_id']) as db:
job=job_access(db,s,job_id)
if archived and s['role']!='owner':fail(403,'Only owners can review archived photographs.')
rows=many(db,'SELECT * FROM job_photos WHERE job_id=? AND customer_id=? AND archived=? ORDER BY position,created_at,id',(job_id,job['customer_id'],int(archived)))
if s['role']!='owner':rows=[p for p in rows if p['uploaded_by']!='public-enquiry' or p['reviewed']]
return [dict(photo_projection(p,s['role']=='owner'),editable=s['role']=='owner' or (p['uploaded_by']==s['subject'] and not p['reviewed']),public_references=active_project_references(db,p['id']) if s['role']=='owner' else []) for p in rows]
@app.post('/api/jobs/{job_id}/photos')
async def upload(job_id:str,request:Request):
s=session(request,csrf=True);raw=await request.body();clean,width,height=decode_image(raw);sha=hashlib.sha256(clean).hexdigest();created=[]
try:
with store.tenant(s['business_id']) as db:
job_access(db,s,job_id) # Recheck scope even when returning a cached retry result.
def action():
job=job_access(db,s,job_id)
prior=one(db,'SELECT p.id,p.archived FROM job_photos p JOIN photo_blobs b ON b.id=p.blob_id WHERE p.job_id=? AND b.sha256=?',(job_id,sha))
if prior:
if prior['archived']:fail(409,'This image is already archived on this job. Ask the owner to restore it.')
return {'id':prior['id'],'duplicate':True}
if db.execute('SELECT COUNT(*) FROM job_photos WHERE job_id=?',(job_id,)).fetchone()[0]>=MAX_JOB_PHOTOS or db.execute('SELECT COUNT(*) FROM job_photos').fetchone()[0]>=MAX_BUSINESS_PHOTOS:fail(409,'The local photo limit is reached: 100 per job and 1,000 per business, including archived photos.')
pid=uid();bid=uid();stamp=now();path=store.photo_path(s['business_id'],bid)
with path.open('xb') as stream:
created.append(path);stream.write(clean);stream.flush();os.fsync(stream.fileno())
db.execute('INSERT INTO photo_blobs VALUES(?,?,?,?,?,?)',(bid,sha,len(clean),width,height,stamp))
db.execute('INSERT INTO job_photos(id,job_id,customer_id,blob_id,uploaded_by,membership_id,position,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)',(pid,job_id,job['customer_id'],bid,s['subject'],s['membership_id'],db.execute('SELECT COALESCE(MAX(position),-1)+1 FROM job_photos WHERE job_id=?',(job_id,)).fetchone()[0],stamp,stamp))
photo_snapshot(db,pid,s,'photo.uploaded')
return {'id':pid,'duplicate':False}
result=once(db,request,{'image_sha256':sha},action)
return result
except BaseException:
for path in created:path.unlink(missing_ok=True)
raise
@app.get('/api/jobs/{job_id}/photos/{photo_id}/image')
def photo_file(job_id:str,photo_id:str,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:photo=photo_access(db,s,job_id,photo_id,archived=True)
return FileResponse(store.photo_path(s['business_id'],photo['blob_id']),media_type='image/png')
@app.put('/api/jobs/{job_id}/photos/{photo_id}')
def edit_photo(job_id:str,photo_id:str,payload:m.PhotoEdit,request:Request):
s=session(request,csrf=True)
try:
with store.tenant(s['business_id']) as db:
photo_access(db,s,job_id,photo_id)
def action():
row=photo_access(db,s,job_id,photo_id)
if s['role']!='owner' and (row['uploaded_by']!=s['subject'] or row['reviewed']):fail(403,'Only the owner can edit a reviewed photo or another person’s submission.')
fields=payload.model_dump(exclude={'version'})
if not db.execute('UPDATE job_photos SET '+','.join(k+'=?' for k in fields)+',reviewed=0,version=version+1,updated_at=? WHERE id=? AND version=?',(*fields.values(),now(),photo_id,payload.version)).rowcount:fail(409,'This photo changed. Reload before saving.')
photo_snapshot(db,photo_id,s,'photo.metadata_updated');return {'version':payload.version+1}
return once(db,request,payload.model_dump(),action)
except sqlite3.IntegrityError:fail(409,'That comparison already has a photo in this phase. Choose another pair name.')
@app.post('/api/jobs/{job_id}/photos/{photo_id}/share')
def share_photo(job_id:str,photo_id:str,payload:m.PhotoShare,request:Request):
s=session(request,csrf=True)
with store.tenant(s['business_id']) as db:
def action():
row=photo_access(db,s,job_id,photo_id)
if payload.shared and not row['reviewed']:fail(409,'Review this photo before sharing it with the customer.')
value=json.dumps({'caption':payload.caption,'alt':payload.alt}) if payload.shared else None
if not db.execute('UPDATE job_photos SET customer_share=?,version=version+1,updated_at=? WHERE id=? AND version=?',(value,now(),photo_id,payload.version)).rowcount:fail(409,'This photo changed. Reload before sharing.')
photo_snapshot(db,photo_id,s,'photo.customer_shared' if payload.shared else 'photo.customer_unshared');return {'version':payload.version+1}
return once(db,request,payload.model_dump(),action)
@app.post('/api/jobs/{job_id}/photos/{photo_id}/state/{action}')
def photo_state(job_id:str,photo_id:str,action:str,payload:m.Version,request:Request):
s=session(request,csrf=True)
if action not in ('review','archive','restore'):fail(404,'Action not available.')
try:
with store.tenant(s['business_id']) as db:
def change():
row=photo_access(db,s,job_id,photo_id,archived=True)
if row['version']!=payload.version:fail(409,'This photo changed. Reload before continuing.')
if action=='archive' and active_project_references(db,photo_id):fail(409,'Withdraw the published projects using this photo before archiving it. Historical images will be retained.')
if action=='review':
if row['archived']:fail(409,'Restore this photo before reviewing it.')
db.execute('UPDATE job_photos SET reviewed=1,version=version+1,updated_at=? WHERE id=?',(now(),photo_id))
else:
wanted=action=='archive'
if bool(row['archived'])==wanted:fail(409,'This photo is already in that state.')
db.execute('UPDATE job_photos SET archived=?,reviewed=0,customer_share=NULL,version=version+1,updated_at=? WHERE id=?',(int(wanted),now(),photo_id))
photo_snapshot(db,photo_id,s,'photo.'+action);return {'version':payload.version+1}
return once(db,request,payload.model_dump(),change)
except sqlite3.IntegrityError:fail(409,'Resolve the existing comparison pair before restoring this photo.')
@app.get('/api/jobs/{job_id}/photos/{photo_id}/history')
def photo_history(job_id:str,photo_id:str,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:
photo_access(db,s,job_id,photo_id,archived=True)
return [{**r,'snapshot':json.loads(r['snapshot'])} for r in many(db,'SELECT * FROM photo_history WHERE photo_id=? ORDER BY id DESC',(photo_id,))]
def customer_job(db,s,job_id):return required(one(db,'SELECT id,customer_id FROM jobs WHERE id=? AND customer_id=?',(job_id,s['subject'])))
@app.get('/api/portal/{slug}/jobs/{job_id}/photos')
def customer_photos(slug:str,job_id:str,request:Request):
s=portal_session(request,slug)
with store.tenant(s['business_id']) as db:
customer_job(db,s,job_id)
return [{'id':r['id'],**json.loads(r['customer_share'])} for r in many(db,'SELECT id,customer_share FROM job_photos WHERE job_id=? AND customer_id=? AND archived=0 AND customer_share IS NOT NULL ORDER BY position,created_at,id',(job_id,s['subject']))]
@app.get('/api/portal/{slug}/jobs/{job_id}/photos/{photo_id}/image')
def customer_photo_file(slug:str,job_id:str,photo_id:str,request:Request):
s=portal_session(request,slug)
with store.tenant(s['business_id']) as db:
customer_job(db,s,job_id)
row=required(one(db,'SELECT blob_id FROM job_photos WHERE id=? AND job_id=? AND customer_id=? AND archived=0 AND customer_share IS NOT NULL',(photo_id,job_id,s['subject'])))
return FileResponse(store.photo_path(s['business_id'],row['blob_id']),media_type='image/png')
@app.get('/api/gallery')
def projects(request:Request,archived:bool=False):
s=session(request)
with store.tenant(s['business_id']) as db:return [decode_project(r) for r in many(db,'SELECT * FROM gallery_projects WHERE archived=? ORDER BY updated_at DESC,id',(int(archived),))]
def decode_project(row):return {**row,'draft':json.loads(row['draft']),'published':json.loads(row['published']) if row['published'] else None}
@app.post('/api/gallery')
def create_project(payload:m.ProjectCreate,request:Request):
s=session(request,csrf=True);content=payload.content.model_dump()
with store.tenant(s['business_id']) as db:
def action():
if db.execute('SELECT COUNT(*) FROM gallery_projects').fetchone()[0]>=MAX_PROJECTS:fail(409,'The local limit of 100 projects including archived projects has been reached.')
if payload.source_job_id:
job_access(db,s,payload.source_job_id)
if db.execute('SELECT 1 FROM gallery_projects WHERE source_job_id=?',(payload.source_job_id,)).fetchone():fail(409,'This job already has a gallery project. Edit or restore that project instead.')
pid=uid();row={'id':pid,'source_job_id':payload.source_job_id}
project_sources(db,row,content)
db.execute('INSERT INTO gallery_projects(id,source_job_id,draft_slug,draft,created_at,updated_at) VALUES(?,?,?,?,?,?)',(pid,payload.source_job_id,content['slug'],json.dumps(content),now(),now()))
project_snapshot(db,pid,s,'project.created');return {'id':pid,'version':1}
return once(db,request,payload.model_dump(),action)
@app.get('/api/gallery/{project_id}')
def read_project(project_id:str,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:
row=project_row(db,project_id,archived=True)
return {**decode_project(row),'history':many(db,'SELECT id,action,actor_id,version,permission_confirmed,created_at FROM gallery_history WHERE project_id=? ORDER BY id DESC',(project_id,))}
@app.put('/api/gallery/{project_id}')
def edit_project(project_id:str,payload:m.ProjectEdit,request:Request):
s=session(request,csrf=True);content=payload.content.model_dump()
with store.tenant(s['business_id']) as db:
def action():
row=project_row(db,project_id);project_sources(db,row,content)
if not db.execute('UPDATE gallery_projects SET draft_slug=?,draft=?,version=version+1,updated_at=? WHERE id=? AND version=?',(content['slug'],json.dumps(content),now(),project_id,payload.version)).rowcount:fail(409,'This project changed. Keep your edits and reload before saving.')
project_snapshot(db,project_id,s,'project.draft_saved');return {'version':payload.version+1}
return once(db,request,payload.model_dump(),action)
@app.post('/api/gallery/{project_id}/publish')
def publish_project(project_id:str,payload:m.ProjectPublish,request:Request):
s=session(request,csrf=True)
with store.tenant(s['business_id']) as db:
def action():
row=project_row(db,project_id)
if row['version']!=payload.version:fail(409,'Save or reload the current draft before publishing.')
content=m.ProjectContent.model_validate_json(row['draft']).model_dump();project_sources(db,row,content,publishing=True)
db.execute('UPDATE gallery_projects SET published=?,version=version+1,updated_at=? WHERE id=?',(json.dumps(content),now(),project_id))
project_snapshot(db,project_id,s,'project.published',True);return {'version':payload.version+1,'slug':content['slug']}
return once(db,request,payload.model_dump(),action)
@app.post('/api/gallery/{project_id}/state/{action}')
def project_state(project_id:str,action:str,payload:m.Version,request:Request):
s=session(request,csrf=True)
if action not in ('withdraw','archive','restore'):fail(404,'Action not available.')
with store.tenant(s['business_id']) as db:
def change():
row=project_row(db,project_id,archived=True)
if row['version']!=payload.version:fail(409,'This project changed. Reload before continuing.')
if action=='withdraw':
if not row['published']:fail(409,'This project is already unpublished.')
db.execute('UPDATE gallery_projects SET published=NULL,version=version+1,updated_at=? WHERE id=?',(now(),project_id))
else:
wanted=action=='archive'
if row['published']:fail(409,'Withdraw this project before archiving it.')
if bool(row['archived'])==wanted:fail(409,'This project is already in that state.')
db.execute('UPDATE gallery_projects SET archived=?,version=version+1,updated_at=? WHERE id=?',(int(wanted),now(),project_id))
project_snapshot(db,project_id,s,'project.'+action);return {'version':payload.version+1}
return once(db,request,payload.model_dump(),change)
@app.post('/api/gallery/{project_id}/restore-draft')
def restore_project(project_id:str,payload:m.ProjectRestore,request:Request):
s=session(request,csrf=True)
with store.tenant(s['business_id']) as db:
def action():
row=project_row(db,project_id)
old=required(one(db,'SELECT * FROM gallery_history WHERE id=? AND project_id=?',(payload.revision_id,project_id)))
saved=json.loads(old['snapshot']);content=m.ProjectContent.model_validate_json(saved['published'] if old['action']=='project.published' else saved['draft']).model_dump()
project_sources(db,row,content)
if not db.execute('UPDATE gallery_projects SET draft=?,draft_slug=?,version=version+1,updated_at=? WHERE id=? AND version=?',(json.dumps(content),content['slug'],now(),project_id,payload.version)).rowcount:fail(409,'This project changed. Reload before restoring.')
project_snapshot(db,project_id,s,'project.restored_to_draft');return {'version':payload.version+1}
return once(db,request,payload.model_dump(),action)
@app.get('/api/gallery/{project_id}/preview')
def preview_project(project_id:str,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:return project_projection(json.loads(project_row(db,project_id)['draft']))
@app.get('/api/gallery/{project_id}/preview/images/{image_id}')
def preview_image(project_id:str,image_id:str,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:
content=json.loads(project_row(db,project_id)['draft']);image=required(next((i for i in content['images'] if i['id']==image_id),None));path=source_path(store,db,s['business_id'],image)
return FileResponse(path,media_type='image/png')
@app.get('/api/gallery/{project_id}/history/{revision_id}')
def history_project(project_id:str,revision_id:int,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:
project_row(db,project_id,archived=True);row=required(one(db,'SELECT * FROM gallery_history WHERE project_id=? AND id=?',(project_id,revision_id)))
return {**row,'snapshot':decode_project(json.loads(row['snapshot']))}
@app.get('/api/gallery/{project_id}/history/{revision_id}/images/{image_id}')
def history_image(project_id:str,revision_id:int,image_id:str,request:Request):
s=session(request)
with store.tenant(s['business_id']) as db:
project_row(db,project_id,archived=True);row=required(one(db,'SELECT * FROM gallery_history WHERE project_id=? AND id=?',(project_id,revision_id)));snapshot=json.loads(row['snapshot'])
content=json.loads(snapshot['published'] if row['action']=='project.published' else snapshot['draft']);image=required(next((i for i in content['images'] if i['id']==image_id),None));path=source_path(store,db,s['business_id'],image)
return FileResponse(path,media_type='image/png')
@app.get('/api/public/{slug}/projects')
def public_index(slug:str):
b=business(slug)
with store.tenant(b['id']) as db:return public_projects(db)
@app.get('/api/public/{slug}/projects/{project_slug}')
def public_detail(slug:str,project_slug:str):
b=business(slug)
with store.tenant(b['id']) as db:
_,content=published_project(db,project_slug);services={s['slug'] for s in public_content(json.loads(db.execute('SELECT published FROM site').fetchone()[0]))['services']};project=project_projection(content,services)
related=[p for p in public_projects(db) if p['slug']!=project_slug and p['category']==project['category']][:3]
return {'project':project,'related':related}
@app.get('/api/public/{slug}/projects/{project_slug}/images/{image_id}')
def public_image(slug:str,project_slug:str,image_id:str):
b=business(slug)
with store.tenant(b['id']) as db:
_,content=published_project(db,project_slug);image=required(next((i for i in content['images'] if i['id']==image_id),None));path=source_path(store,db,b['id'],image)
return FileResponse(path,media_type='image/png')
if static_dir:
@app.get('/site/{slug}/work/projects')
@app.get('/site/{slug}/work/projects/{project_slug}')
def public_shell(slug:str,request:Request,project_slug:str|None=None):
b=business(slug)
with store.tenant(b['id']) as db:
site=public_content(json.loads(db.execute('SELECT published FROM site').fetchone()[0]));projects=public_projects(db)
project=project_projection(published_project(db,project_slug)[1]) if project_slug else None
return HTMLResponse(gallery_shell(static_dir,site,project,'http://'+request.headers['host']+'/site/'+slug+'/work/projects',projects))