From 61c64add5bb5228ca768654ae6473a2234168b4d Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Wed, 8 Oct 2025 21:00:44 +0200 Subject: [PATCH 01/19] back up --- .gitignore | 8 +++ api/errors.py | 10 ++++ api/main.py | 54 ++++++++++++++++++++ api/models.py | 21 ++++++++ worker/main.py | 79 +++++++++++++++++++++++++++++ worker/sift.py | 134 +++++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 306 insertions(+) create mode 100644 .gitignore create mode 100644 api/errors.py create mode 100644 api/main.py create mode 100644 api/models.py create mode 100644 worker/main.py create mode 100644 worker/sift.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d990c3c --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +api/__pycache__/* +worker/__pycache__ +senior_ai_engineer_assignment/* +results/* +test/* +.env +.envrc +adc.json \ No newline at end of file diff --git a/api/errors.py b/api/errors.py new file mode 100644 index 0000000..8c42f10 --- /dev/null +++ b/api/errors.py @@ -0,0 +1,10 @@ +from fastapi import HTTPException +from api.main import FileType + + +class InvalidFileTypeException(HTTPException): + def __init__(self, type: FileType): + super().__init__( + status_code=400, + detail=f"Invalid upload type, must be {FileType.VIDEO} or {FileType.IMAGE}, got {type}", + ) diff --git a/api/main.py b/api/main.py new file mode 100644 index 0000000..2674f72 --- /dev/null +++ b/api/main.py @@ -0,0 +1,54 @@ +from fastapi import FastAPI +from google.cloud import storage + +from datetime import datetime, timedelta, timezone + +from api.errors import InvalidFileTypeException +from api.models import UploadResponse, ResultsResponse, FileType + +api = FastAPI() +BUCKET_NAME = "code_challenge_logo_detection" +storage_client = storage.Client(project="jing") + + +@api.get("/health") +async def health(): + return {"status": "ok"} + + +@api.post("/upload", response_model=UploadResponse) +async def upload(file_name: str, type: FileType): + if not type.is_valid(): + raise InvalidFileTypeException(type) + + blob = get_upload_blob(file_name, type) + expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=15) + signed_url = blob.generate_signed_url( + expiry=expiry, + method="PUT", + ) + return UploadResponse(signed_url=signed_url, expiry=expiry) + + +@api.get("/results", response_model=ResultsResponse) +async def get_results(file_name: str, file_type: FileType): + if not file_type.is_valid(): + raise InvalidFileTypeException(file_type) + + blob = get_results_blob(file_name, file_type) + expiry = datetime.now(tz=timezone.utc) + timedelta(hours=1) + return ResultsResponse( + signed_url=blob.generate_signed_url(expiry=expiry, method="GET") + ) + + +def get_upload_blob(file_name: str, file_type: FileType) -> storage.Blob: + return storage_client.bucket(BUCKET_NAME).blob( + f"results/{file_type.value}/{file_name}" + ) + + +def get_results_blob(file_name: str, file_type: FileType) -> storage.Blob: + return storage_client.bucket(BUCKET_NAME).blob( + f"results/{file_type.value}/{file_name}" + ) diff --git a/api/models.py b/api/models.py new file mode 100644 index 0000000..6cef02c --- /dev/null +++ b/api/models.py @@ -0,0 +1,21 @@ +from pydantic import BaseModel +from datetime import datetime +from enum import Enum + + +class UploadResponse(BaseModel): + signed_url: str + expiry: datetime + + +class ResultsResponse(BaseModel): + signed_url: str + expiry: datetime + + +class FileType(str, Enum): + VIDEO = "video" + IMAGE = "image" + + def is_valid(self) -> bool: + return self == FileType.VIDEO or self == FileType.IMAGE diff --git a/worker/main.py b/worker/main.py new file mode 100644 index 0000000..98a5824 --- /dev/null +++ b/worker/main.py @@ -0,0 +1,79 @@ +from typing import Dict + +import functions_framework +from flask import Request +from google.cloud import storage +from pydantic import BaseModel, ValidationError + +from sift import detect_logo_with_sift + +BUCKET_NAME = "code_challenge_logo_detection" +storage_client = storage.Client(project="jing") + + +class ProcessVideoRequest(BaseModel): + video_gcs_path: str + logo_gcs_path: str + output_gcs_path: str + + +@functions_framework.http +def process_video(request: Request): + + data: Dict = request.get_json(force=True) + try: + request = ProcessVideoRequest(**data) + except ValidationError as e: + return {"error": "Invalid request", "details": e.errors()}, 400 + + video_path = request.video_gcs_path + logo_path = request.logo_gcs_path + output_path = request.output_gcs_path + + bucket = storage_client.bucket(BUCKET_NAME) + + local_video_path = get_local_path(video_path) + local_logo_path = get_local_path(logo_path) + local_output_path = get_local_path(output_path) + + video_blob = bucket.blob(video_path) + video_blob.download_to_filename(local_video_path) + + logo_blob = bucket.blob(logo_path) + logo_blob.download_to_filename(local_logo_path) + + detect_logo_with_sift( + logo_path=local_logo_path, + video_path=local_video_path, + output_path=local_output_path, + min_matches=15, + ratio_threshold=0.75, + ) + + output_blob = bucket.blob(output_path) + output_blob.upload_from_filename(output_path) + + return {"result": output_blob.path} + + +def get_local_path(path: str) -> str: + return f"/tmp/{path}" + + +# if __name__ == "__main__": +# from werkzeug.test import EnvironBuilder + + +# builder = EnvironBuilder( +# method="POST", +# path="/process", +# json={ +# "video_gcs_path": "Video_1.mp4", +# "logo_gcs_path": "neurons_logo.jpg", +# "output_gcs_path": "output_video_1_fearure_matching_SIFT.mp4", +# }, +# ) +# env = builder.get_environ() +# req = Request(env) +# response = process_video(req) +# print(response) diff --git a/worker/sift.py b/worker/sift.py new file mode 100644 index 0000000..4511beb --- /dev/null +++ b/worker/sift.py @@ -0,0 +1,134 @@ +import cv2 +import numpy as np + + +def detect_logo_with_sift( + logo_path: str, + video_path: str, + output_path: str, + min_matches: int = 10, + ratio_threshold: float = 0.75, +): + + logo = cv2.imread(logo_path) + if logo is None: + raise ValueError(f"Could not load logo: {logo_path}") + + logo_gray = cv2.cvtColor(logo, cv2.COLOR_BGR2GRAY) + + detector = cv2.SIFT_create() + + key_point, descriptor = detector.detectAndCompute(logo_gray, None) + + if descriptor is None or len(key_point) < 4: + raise ValueError("Not enough features in logo. Try a more detailed logo image.") + + FLANN_INDEX_KDTREE = 1 + index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) + search_params = dict(checks=50) + matcher = cv2.FlannBasedMatcher(index_params, search_params) + + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + raise ValueError(f"Could not open video: {video_path}") + + fps, width, height, total_frames = get_video_info(cap) + + fourcc = cv2.VideoWriter_fourcc(*"mp4v") + out = cv2.VideoWriter(output_path, fourcc, fps, (width, height)) + + frame_count = 0 + frames_with_logo = 0 + total_matches = 0 + + while True: + ret, frame = cap.read() + if not ret: + break + + frame_count += 1 + frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + + frame_kp, frame_desc = detector.detectAndCompute(frame_gray, None) + + if frame_desc is not None and len(frame_kp) >= 4: + matches = matcher.knnMatch(descriptor, frame_desc, k=2) + + good_matches = [] + for match_pair in matches: + if len(match_pair) == 2: + m, n = match_pair + if m.distance < ratio_threshold * n.distance: + good_matches.append(m) + + if len(good_matches) >= min_matches: + frames_with_logo += 1 + total_matches += len(good_matches) + + src_pts = np.float32( + [key_point[m.queryIdx].pt for m in good_matches] + ).reshape(-1, 1, 2) + + dst_pts = np.float32( + [frame_kp[m.trainIdx].pt for m in good_matches] + ).reshape(-1, 1, 2) + + M, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) + + if M is not None: + h, w = logo_gray.shape + logo_corners = np.float32([[0, 0], [w, 0], [w, h], [0, h]]).reshape( + -1, 1, 2 + ) + + frame_corners = cv2.perspectiveTransform(logo_corners, M) + + frame = cv2.polylines( + frame, [np.int32(frame_corners)], True, (0, 255, 0), 3 + ) + + x, y = int(frame_corners[0][0][0]), int(frame_corners[0][0][1]) + label = f"Logo: {len(good_matches)} matches" + cv2.putText( + frame, + label, + (x, max(y - 10, 20)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 255, 0), + 2, + ) + + out.write(frame) + + if frame_count % 30 == 0: + progress = (frame_count / total_frames) * 100 + print(f"Progress: {progress:.1f}% ({frame_count}/{total_frames})") + + cap.release() + out.release() + + +def get_video_info(cap: cv2.VideoCapture) -> tuple[int, int, int, int]: + fps = int(cap.get(cv2.CAP_PROP_FPS)) + width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + + print(f"\nVideo: {width}x{height}, {fps} FPS, {total_frames} frames") + return fps, width, height, total_frames + + +if __name__ == "__main__": + + logo_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/neurons_logo.jpg" + video_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/Video_1.mp4" + output_path = "/Users/jing/code_challenge/video-logo-tracker/output_video_1_fearure_matching_SIFT.mp4" + + detect_logo_with_sift( + logo_path=logo_path, + video_path=video_path, + output_path=output_path, + min_matches=15, + ratio_threshold=0.75, + ) From 45cf339f9c685ad6f24ba4b23e79be755f4082fc Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Thu, 9 Oct 2025 21:59:37 +0200 Subject: [PATCH 02/19] back up --- .gitignore | 4 +- api/db.py | 118 +++++++++++++++++++++++++++++++++ api/errors.py | 39 +++++++++-- api/main.py | 173 +++++++++++++++++++++++++++++++++++++++++-------- api/models.py | 47 +++++++++++--- worker/sift.py | 4 +- 6 files changed, 339 insertions(+), 46 deletions(-) create mode 100644 api/db.py diff --git a/.gitignore b/.gitignore index d990c3c..b6e861d 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,6 @@ results/* test/* .env .envrc -adc.json \ No newline at end of file +adc.json +.DS_Store +logo_detection.db \ No newline at end of file diff --git a/api/db.py b/api/db.py new file mode 100644 index 0000000..018af27 --- /dev/null +++ b/api/db.py @@ -0,0 +1,118 @@ +import aiosqlite +from models import JobStatus, Job +import logging +from typing import List, Optional + +DB_PATH = "logo_detection.db" + + +async def init_db(): + statuses = ", ".join([f"'{status.value}'" for status in JobStatus]) + query = f""" + CREATE TABLE IF NOT EXISTS logo_detection ( + id TEXT PRIMARY KEY, + video_gcs_uri TEXT NOT NULL, + logo_gcs_uri TEXT NOT NULL, + output_gcs_uri TEXT, + status TEXT CHECK(status IN ({statuses})), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + logging.info(query) + async with aiosqlite.connect(DB_PATH) as db: + await db.execute(query) + await db.commit() + + +async def drop_db(): + async with aiosqlite.connect(DB_PATH) as db: + await db.execute("DROP TABLE IF EXISTS logo_detection") + await db.commit() + + +async def get(job_id: str) -> Job: + async with aiosqlite.connect(DB_PATH) as db: + async with db.execute( + "SELECT id, video_gcs_uri, logo_gcs_uri, output_gcs_uri, status, error, created_at, updated_at FROM logo_detection WHERE id = ?", + (job_id,), + ) as cursor: + row = await cursor.fetchone() + if row is None: + return None + return Job( + id=row[0], + video_gcs_uri=row[1], + logo_gcs_uri=row[2], + output_gcs_uri=row[3], + status=row[4], + error=row[5], + created_at=row[6], + updated_at=row[7], + ) + + +async def save(job: Job): + async with aiosqlite.connect(DB_PATH) as db: + await db.execute( + """INSERT INTO + logo_detection (id, + video_gcs_uri, + logo_gcs_uri, + output_gcs_uri, + status, + error, + created_at, + updated_at) + VALUES + (:id, :video_gcs_uri, :logo_gcs_uri, :output_gcs_uri, :status, :error, :created_at, :updated_at) + ON + CONFLICT(id) DO + UPDATE + SET + video_gcs_uri=:video_gcs_uri, + logo_gcs_uri=:logo_gcs_uri, + output_gcs_uri=:output_gcs_uri, + status=:status, + error=:error, + updated_at=CURRENT_TIMESTAMP + WHERE id=:id + """, + { + "id": job.id, + "video_gcs_uri": job.video_gcs_uri, + "logo_gcs_uri": job.logo_gcs_uri, + "output_gcs_uri": job.output_gcs_uri, + "status": job.status, + "error": job.error, + "created_at": job.created_at, + "updated_at": job.updated_at, + }, + ) + await db.commit() + + +async def delete(job_id: str): + async with aiosqlite.connect(DB_PATH) as db: + await db.execute("DELETE FROM logo_detection WHERE id = ?", (job_id,)) + await db.commit() + + +async def list() -> List[Job]: + async with aiosqlite.connect(DB_PATH) as db: + async with db.execute( + "SELECT id, video_gcs_uri, logo_gcs_uri, output_gcs_uri, status, created_at, updated_at FROM logo_detection" + ) as cursor: + rows = await cursor.fetchall() + return [ + Job( + id=row[0], + video_gcs_uri=row[1], + logo_gcs_uri=row[2], + output_gcs_uri=row[3], + status=row[4], + created_at=row[5], + updated_at=row[6], + ) + for row in rows + ] diff --git a/api/errors.py b/api/errors.py index 8c42f10..3b428ad 100644 --- a/api/errors.py +++ b/api/errors.py @@ -1,10 +1,41 @@ from fastapi import HTTPException -from api.main import FileType -class InvalidFileTypeException(HTTPException): - def __init__(self, type: FileType): +class InvalidVideoFileExtensionException(HTTPException): + def __init__(self, file_name: str): super().__init__( status_code=400, - detail=f"Invalid upload type, must be {FileType.VIDEO} or {FileType.IMAGE}, got {type}", + detail=f"Invalid video file extension, must be .mp4 got {file_name}", + ) + + +class InvalidImageFileExtensionException(HTTPException): + def __init__(self, file_name: str): + super().__init__( + status_code=400, + detail=f"Invalid image file extension, must be .jpg got {file_name}", + ) + + +class MissingVideoFileException(HTTPException): + def __init__(self): + super().__init__( + status_code=400, + detail="Missing video file", + ) + + +class MissingLogoFileException(HTTPException): + def __init__(self): + super().__init__( + status_code=400, + detail="Missing logo file", + ) + + +class JobNotFoundException(HTTPException): + def __init__(self, job_id: str): + super().__init__( + status_code=404, + detail=f"Job not found with id {job_id}", ) diff --git a/api/main.py b/api/main.py index 2674f72..8258951 100644 --- a/api/main.py +++ b/api/main.py @@ -1,14 +1,55 @@ -from fastapi import FastAPI +from fastapi import FastAPI, Request +from contextlib import asynccontextmanager from google.cloud import storage +from typing import List +import logging +from logging import getLogger, INFO from datetime import datetime, timedelta, timezone +import uuid + +from errors import ( + InvalidVideoFileExtensionException, + InvalidImageFileExtensionException, + JobNotFoundException, + MissingVideoFileException, + MissingLogoFileException, +) +from models import ( + FileType, + Job, + JobStatus, + SignedURL, + UploadRequest, + JobUploadURLs, + JobPatchRequest, +) +from db import ( + get, + save, + delete, + list, + init_db, + drop_db, +) + +LOGGER = getLogger(__name__) +logging.basicConfig(level=INFO) +DB_PATH = "logo_detection.db" + + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + LOGGER.info("database initialized") + yield + await drop_db() + LOGGER.info("database dropped") -from api.errors import InvalidFileTypeException -from api.models import UploadResponse, ResultsResponse, FileType -api = FastAPI() BUCKET_NAME = "code_challenge_logo_detection" storage_client = storage.Client(project="jing") +api = FastAPI(lifespan=lifespan) @api.get("/health") @@ -16,39 +57,115 @@ async def health(): return {"status": "ok"} -@api.post("/upload", response_model=UploadResponse) -async def upload(file_name: str, type: FileType): - if not type.is_valid(): - raise InvalidFileTypeException(type) - - blob = get_upload_blob(file_name, type) - expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=15) - signed_url = blob.generate_signed_url( - expiry=expiry, - method="PUT", +@api.post("/job", response_model=Job) +async def create_job(request: UploadRequest) -> Job: + validate_upload_requests(request) + job_id = str(uuid.uuid4()) + upload_urls = generate_upload_urls(request, job_id) + job = Job( + id=job_id, + video_gcs_uri=upload_urls.video_file.url, + logo_gcs_uri=upload_urls.logo_file.url, + output_gcs_uri=None, + status=JobStatus.UPLOADING, + created_at=datetime.now(tz=timezone.utc), + updated_at=datetime.now(tz=timezone.utc), ) - return UploadResponse(signed_url=signed_url, expiry=expiry) + await save(job) + return job -@api.get("/results", response_model=ResultsResponse) -async def get_results(file_name: str, file_type: FileType): - if not file_type.is_valid(): - raise InvalidFileTypeException(file_type) +@api.get("/job/{job_id}", response_model=Job) +async def get_job(job_id: str) -> Job: + job = await get(job_id) + if job is None: + raise JobNotFoundException(job_id) + return job - blob = get_results_blob(file_name, file_type) - expiry = datetime.now(tz=timezone.utc) + timedelta(hours=1) - return ResultsResponse( - signed_url=blob.generate_signed_url(expiry=expiry, method="GET") - ) + +@api.post("/job/{job_id}/run", response_model=Job) +async def run_job(job_id: str) -> Job: + # TODO trigger cloud function + + job = await get(job_id) + if job is None: + raise JobNotFoundException(job_id) + job.status = JobStatus.RUNNING + await save(job) + return await get(job_id) + + +@api.get("/jobs", response_model=List[Job]) +async def list_jobs() -> List[Job]: + return await list() + + +@api.patch("/job/{job_id}", response_model=Job) +async def job_done(job_id: str, request: JobPatchRequest) -> Job: + job = await get(job_id) + if job is None: + raise JobNotFoundException(job_id) + job.status = request.status + # TODO verify output_gcs_uri + if request.output_gcs_uri: + job.output_gcs_uri = request.output_gcs_uri + if request.error: + job.error = request.error + if request.status: + job.status = request.status + await save(job) + return await get(job_id) -def get_upload_blob(file_name: str, file_type: FileType) -> storage.Blob: +@api.delete("/job/{job_id}", status_code=204) +async def delete_job(job_id: str): + await delete(job_id) + return + + +def get_upload_blob(file_name: str, file_type: FileType, job_id: str) -> storage.Blob: return storage_client.bucket(BUCKET_NAME).blob( - f"results/{file_type.value}/{file_name}" + f"uploads/{file_type.value}/{job_id}/{file_name}" ) -def get_results_blob(file_name: str, file_type: FileType) -> storage.Blob: +def get_result_blob(file_name: str, file_type: FileType, job_id: str) -> storage.Blob: return storage_client.bucket(BUCKET_NAME).blob( - f"results/{file_type.value}/{file_name}" + f"results/{file_type.value}/{job_id}/{file_name}" + ) + + +def validate_upload_requests(request: UploadRequest): + video_file_name = request.video_file_name + logo_file_name = request.logo_file_name + if video_file_name is None or video_file_name == "": + raise MissingVideoFileException() + if logo_file_name is None or logo_file_name == "": + raise MissingLogoFileException() + + if not video_file_name.lower().endswith((".mp4")): + raise InvalidVideoFileExtensionException(video_file_name) + if not logo_file_name.lower().endswith((".jpg")): + raise InvalidImageFileExtensionException(logo_file_name) + return + + +def generate_upload_urls( + request: UploadRequest, + job_id: str, +) -> JobUploadURLs: + expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=15) + video_blob = get_upload_blob(request.video_file_name, FileType.VIDEO, job_id) + logo_blob = get_upload_blob(request.logo_file_name, FileType.IMAGE, job_id) + video_gcs_uri = video_blob.generate_signed_url( + expiration=expiry, + method="PUT", + ) + logo_gcs_uri = logo_blob.generate_signed_url( + expiration=expiry, + method="PUT", + ) + return JobUploadURLs( + video_file=SignedURL(url=video_gcs_uri, expiry=expiry), + logo_file=SignedURL(url=logo_gcs_uri, expiry=expiry), ) diff --git a/api/models.py b/api/models.py index 6cef02c..6aba4b7 100644 --- a/api/models.py +++ b/api/models.py @@ -1,21 +1,48 @@ from pydantic import BaseModel from datetime import datetime from enum import Enum +from typing import Optional -class UploadResponse(BaseModel): - signed_url: str - expiry: datetime +class FileType(str, Enum): + VIDEO = "video" + IMAGE = "image" -class ResultsResponse(BaseModel): - signed_url: str +class SignedURL(BaseModel): + url: str expiry: datetime -class FileType(str, Enum): - VIDEO = "video" - IMAGE = "image" +class JobUploadURLs(BaseModel): + video_file: SignedURL + logo_file: SignedURL + + +class UploadRequest(BaseModel): + video_file_name: str + logo_file_name: str + + +class Job(BaseModel): + id: str + video_gcs_uri: str + logo_gcs_uri: str + output_gcs_uri: Optional[str] + status: str + error: Optional[str] + created_at: datetime + updated_at: datetime + + +class JobStatus(str, Enum): + UPLOADING = "uploading" + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + - def is_valid(self) -> bool: - return self == FileType.VIDEO or self == FileType.IMAGE +class JobPatchRequest(BaseModel): + output_gcs_uri: Optional[str] + status: JobStatus + error: Optional[str] diff --git a/worker/sift.py b/worker/sift.py index 4511beb..d9165d6 100644 --- a/worker/sift.py +++ b/worker/sift.py @@ -6,7 +6,7 @@ def detect_logo_with_sift( logo_path: str, video_path: str, output_path: str, - min_matches: int = 10, + min_matches: int = 15, ratio_threshold: float = 0.75, ): @@ -129,6 +129,4 @@ def get_video_info(cap: cv2.VideoCapture) -> tuple[int, int, int, int]: logo_path=logo_path, video_path=video_path, output_path=output_path, - min_matches=15, - ratio_threshold=0.75, ) From 6f1f105f976c71b5018468a64560801db365fb51 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Fri, 10 Oct 2025 20:12:41 +0200 Subject: [PATCH 03/19] back up --- api/input.py | 26 +++++++ api/main.py | 149 ++++++++++++--------------------------- api/models.py | 25 +------ api/output.py | 49 +++++++++++++ api/{ => services}/db.py | 70 ++++++++++++++---- api/services/storage.py | 45 ++++++++++++ 6 files changed, 223 insertions(+), 141 deletions(-) create mode 100644 api/input.py create mode 100644 api/output.py rename api/{ => services}/db.py (69%) create mode 100644 api/services/storage.py diff --git a/api/input.py b/api/input.py new file mode 100644 index 0000000..955b1aa --- /dev/null +++ b/api/input.py @@ -0,0 +1,26 @@ +from pydantic import BaseModel +from typing import Optional +from models import JobStatus +from services.db import JobDB + + +class UploadInput(BaseModel): + video_file_name: str + logo_file_name: str + + +class JobPatchInput(BaseModel): + output_gcs_uri: Optional[str] + status: JobStatus + error: Optional[str] + + def to_job_db(self, job: JobDB) -> JobDB: + job.status = self.status + if self.output_gcs_uri: + job.output_gcs_uri = self.output_gcs_uri + if self.error: + job.error = self.error + if self.status: + job.status = self.status + + return job diff --git a/api/main.py b/api/main.py index 8258951..17184c3 100644 --- a/api/main.py +++ b/api/main.py @@ -1,11 +1,9 @@ -from fastapi import FastAPI, Request +from fastapi import FastAPI from contextlib import asynccontextmanager -from google.cloud import storage from typing import List -import logging from logging import getLogger, INFO -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone import uuid from errors import ( @@ -15,27 +13,15 @@ MissingVideoFileException, MissingLogoFileException, ) -from models import ( - FileType, - Job, - JobStatus, - SignedURL, - UploadRequest, - JobUploadURLs, - JobPatchRequest, -) -from db import ( - get, - save, - delete, - list, - init_db, - drop_db, -) +from models import JobStatus +from input import UploadInput, JobPatchInput +from output import JobOutput + +from services.db import get, save, delete, list, init_db, drop_db, JobDB +from services.storage import generate_upload_urls LOGGER = getLogger(__name__) -logging.basicConfig(level=INFO) -DB_PATH = "logo_detection.db" +LOGGER.setLevel(INFO) @asynccontextmanager @@ -47,8 +33,6 @@ async def lifespan(app: FastAPI): LOGGER.info("database dropped") -BUCKET_NAME = "code_challenge_logo_detection" -storage_client = storage.Client(project="jing") api = FastAPI(lifespan=lifespan) @@ -57,87 +41,69 @@ async def health(): return {"status": "ok"} -@api.post("/job", response_model=Job) -async def create_job(request: UploadRequest) -> Job: - validate_upload_requests(request) - job_id = str(uuid.uuid4()) - upload_urls = generate_upload_urls(request, job_id) - job = Job( - id=job_id, - video_gcs_uri=upload_urls.video_file.url, - logo_gcs_uri=upload_urls.logo_file.url, - output_gcs_uri=None, +@api.post("/jobs", response_model=JobOutput) +async def create_job(input: UploadInput) -> JobOutput: + validate_upload_input(input) + id = str(uuid.uuid4()) + upload_urls = generate_upload_urls(input, id) + job = JobDB( + id=id, + video_gcs_uri=upload_urls.video_file.uri, + logo_gcs_uri=upload_urls.logo_file.uri, status=JobStatus.UPLOADING, created_at=datetime.now(tz=timezone.utc), - updated_at=datetime.now(tz=timezone.utc), ) await save(job) - return job + saved_job = await get(id) + return JobOutput.from_job_db(saved_job) -@api.get("/job/{job_id}", response_model=Job) -async def get_job(job_id: str) -> Job: - job = await get(job_id) +@api.get("/jobs/{id}", response_model=JobOutput) +async def get_job(id: str) -> JobOutput: + job = await get(id) if job is None: - raise JobNotFoundException(job_id) - return job + raise JobNotFoundException(id) + return JobOutput.from_job_db(job) -@api.post("/job/{job_id}/run", response_model=Job) -async def run_job(job_id: str) -> Job: +@api.post("/jobs/{id}/run", response_model=JobOutput) +async def run_job(id: str) -> JobOutput: # TODO trigger cloud function - job = await get(job_id) + job = await get(id) if job is None: - raise JobNotFoundException(job_id) + raise JobNotFoundException(id) job.status = JobStatus.RUNNING + job.error = None + job.output_gcs_uri = None await save(job) - return await get(job_id) + return JobOutput.from_job_db(job) -@api.get("/jobs", response_model=List[Job]) -async def list_jobs() -> List[Job]: - return await list() +@api.get("/jobs", response_model=List[JobOutput]) +async def list_jobs() -> List[JobOutput]: + return JobOutput.from_jobs_db(await list()) -@api.patch("/job/{job_id}", response_model=Job) -async def job_done(job_id: str, request: JobPatchRequest) -> Job: - job = await get(job_id) +@api.patch("/jobs/{id}", response_model=JobOutput) +async def job_done(id: str, input: JobPatchInput) -> JobOutput: + job = await get(id) if job is None: - raise JobNotFoundException(job_id) - job.status = request.status - # TODO verify output_gcs_uri - if request.output_gcs_uri: - job.output_gcs_uri = request.output_gcs_uri - if request.error: - job.error = request.error - if request.status: - job.status = request.status + raise JobNotFoundException(id) + job = input.to_job_db(job) await save(job) - return await get(job_id) + return JobOutput.from_job_db(job) -@api.delete("/job/{job_id}", status_code=204) -async def delete_job(job_id: str): - await delete(job_id) +@api.delete("/jobs/{id}", status_code=204) +async def delete_job(id: str): + await delete(id) return -def get_upload_blob(file_name: str, file_type: FileType, job_id: str) -> storage.Blob: - return storage_client.bucket(BUCKET_NAME).blob( - f"uploads/{file_type.value}/{job_id}/{file_name}" - ) - - -def get_result_blob(file_name: str, file_type: FileType, job_id: str) -> storage.Blob: - return storage_client.bucket(BUCKET_NAME).blob( - f"results/{file_type.value}/{job_id}/{file_name}" - ) - - -def validate_upload_requests(request: UploadRequest): - video_file_name = request.video_file_name - logo_file_name = request.logo_file_name +def validate_upload_input(input: UploadInput): + video_file_name = input.video_file_name + logo_file_name = input.logo_file_name if video_file_name is None or video_file_name == "": raise MissingVideoFileException() if logo_file_name is None or logo_file_name == "": @@ -148,24 +114,3 @@ def validate_upload_requests(request: UploadRequest): if not logo_file_name.lower().endswith((".jpg")): raise InvalidImageFileExtensionException(logo_file_name) return - - -def generate_upload_urls( - request: UploadRequest, - job_id: str, -) -> JobUploadURLs: - expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=15) - video_blob = get_upload_blob(request.video_file_name, FileType.VIDEO, job_id) - logo_blob = get_upload_blob(request.logo_file_name, FileType.IMAGE, job_id) - video_gcs_uri = video_blob.generate_signed_url( - expiration=expiry, - method="PUT", - ) - logo_gcs_uri = logo_blob.generate_signed_url( - expiration=expiry, - method="PUT", - ) - return JobUploadURLs( - video_file=SignedURL(url=video_gcs_uri, expiry=expiry), - logo_file=SignedURL(url=logo_gcs_uri, expiry=expiry), - ) diff --git a/api/models.py b/api/models.py index 6aba4b7..e0c294c 100644 --- a/api/models.py +++ b/api/models.py @@ -1,7 +1,6 @@ from pydantic import BaseModel from datetime import datetime from enum import Enum -from typing import Optional class FileType(str, Enum): @@ -10,7 +9,7 @@ class FileType(str, Enum): class SignedURL(BaseModel): - url: str + uri: str expiry: datetime @@ -19,30 +18,8 @@ class JobUploadURLs(BaseModel): logo_file: SignedURL -class UploadRequest(BaseModel): - video_file_name: str - logo_file_name: str - - -class Job(BaseModel): - id: str - video_gcs_uri: str - logo_gcs_uri: str - output_gcs_uri: Optional[str] - status: str - error: Optional[str] - created_at: datetime - updated_at: datetime - - class JobStatus(str, Enum): UPLOADING = "uploading" RUNNING = "running" SUCCESS = "success" FAILED = "failed" - - -class JobPatchRequest(BaseModel): - output_gcs_uri: Optional[str] - status: JobStatus - error: Optional[str] diff --git a/api/output.py b/api/output.py new file mode 100644 index 0000000..c3e9019 --- /dev/null +++ b/api/output.py @@ -0,0 +1,49 @@ +from pydantic import BaseModel +from typing import Optional, List +from datetime import datetime +from models import JobStatus +from db import JobDB +from pathlib import Path +from pydantic import computed_field +from urllib.parse import urlparse +from logging import getLogger + +LOGGER = getLogger(__name__) + + +class GCSObject(BaseModel): + uri: str + + @computed_field + @property + def name(self) -> str: + parsed = urlparse(self.uri) + return Path(parsed.path).name + + +class JobOutput(BaseModel): + id: str + video: GCSObject + logo: GCSObject + output: Optional[GCSObject] + status: JobStatus + error: Optional[str] + created_at: datetime + updated_at: datetime + + def from_job_db(job_db: JobDB) -> "JobOutput": + return JobOutput( + id=job_db.id, + video=GCSObject(uri=job_db.video_gcs_uri), + logo=GCSObject(uri=job_db.logo_gcs_uri), + output=( + GCSObject(uri=job_db.output_gcs_uri) if job_db.output_gcs_uri else None + ), + status=job_db.status, + error=job_db.error, + created_at=job_db.created_at, + updated_at=job_db.updated_at, + ) + + def from_jobs_db(jobs_db: List[JobDB]) -> List["JobOutput"]: + return [JobOutput.from_job_db(job_db) for job_db in jobs_db] diff --git a/api/db.py b/api/services/db.py similarity index 69% rename from api/db.py rename to api/services/db.py index 018af27..fdbafef 100644 --- a/api/db.py +++ b/api/services/db.py @@ -1,11 +1,23 @@ import aiosqlite -from models import JobStatus, Job -import logging +from models import JobStatus from typing import List, Optional +from datetime import datetime, timezone +from pydantic import BaseModel DB_PATH = "logo_detection.db" +class JobDB(BaseModel): + id: str + video_gcs_uri: str + logo_gcs_uri: str + output_gcs_uri: Optional[str] = None + status: JobStatus + error: Optional[str] = None + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + async def init_db(): statuses = ", ".join([f"'{status.value}'" for status in JobStatus]) query = f""" @@ -15,11 +27,11 @@ async def init_db(): logo_gcs_uri TEXT NOT NULL, output_gcs_uri TEXT, status TEXT CHECK(status IN ({statuses})), + error TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ - logging.info(query) async with aiosqlite.connect(DB_PATH) as db: await db.execute(query) await db.commit() @@ -31,16 +43,30 @@ async def drop_db(): await db.commit() -async def get(job_id: str) -> Job: +async def get(id: str) -> JobDB: async with aiosqlite.connect(DB_PATH) as db: async with db.execute( - "SELECT id, video_gcs_uri, logo_gcs_uri, output_gcs_uri, status, error, created_at, updated_at FROM logo_detection WHERE id = ?", - (job_id,), + """ + SELECT + id, + video_gcs_uri, + logo_gcs_uri, + output_gcs_uri, + status, + error, + created_at, + updated_at + FROM + logo_detection + WHERE + id = :id + """, + {"id": id}, ) as cursor: row = await cursor.fetchone() if row is None: return None - return Job( + return JobDB( id=row[0], video_gcs_uri=row[1], logo_gcs_uri=row[2], @@ -52,7 +78,8 @@ async def get(job_id: str) -> Job: ) -async def save(job: Job): +async def save(job: JobDB): + job.updated_at = datetime.now(tz=timezone.utc) async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO @@ -75,7 +102,7 @@ async def save(job: Job): output_gcs_uri=:output_gcs_uri, status=:status, error=:error, - updated_at=CURRENT_TIMESTAMP + updated_at=:updated_at WHERE id=:id """, { @@ -92,26 +119,39 @@ async def save(job: Job): await db.commit() -async def delete(job_id: str): +async def delete(id: str): async with aiosqlite.connect(DB_PATH) as db: - await db.execute("DELETE FROM logo_detection WHERE id = ?", (job_id,)) + await db.execute("DELETE FROM logo_detection WHERE id = :id", {"id": id}) await db.commit() -async def list() -> List[Job]: +async def list() -> List[JobDB]: async with aiosqlite.connect(DB_PATH) as db: async with db.execute( - "SELECT id, video_gcs_uri, logo_gcs_uri, output_gcs_uri, status, created_at, updated_at FROM logo_detection" + """ + SELECT + id, + video_gcs_uri, + logo_gcs_uri, + output_gcs_uri, + status, + error, + created_at, + updated_at + FROM + logo_detection + """ ) as cursor: rows = await cursor.fetchall() return [ - Job( + JobDB( id=row[0], video_gcs_uri=row[1], logo_gcs_uri=row[2], output_gcs_uri=row[3], status=row[4], - created_at=row[5], + error=row[5], + created_at=row[6], updated_at=row[6], ) for row in rows diff --git a/api/services/storage.py b/api/services/storage.py new file mode 100644 index 0000000..b9dbcb6 --- /dev/null +++ b/api/services/storage.py @@ -0,0 +1,45 @@ +from google.cloud import storage +from datetime import datetime, timedelta, timezone +from input import UploadInput +from models import ( + FileType, + SignedURL, + JobUploadURLs, +) + + +BUCKET_NAME = "code_challenge_logo_detection" +storage_client = storage.Client(project="jing") + + +def get_upload_blob(file_name: str, file_type: FileType, id: str) -> storage.Blob: + return storage_client.bucket(BUCKET_NAME).blob( + f"uploads/{file_type.value}/{id}/{file_name}" + ) + + +def get_result_blob(file_name: str, file_type: FileType, id: str) -> storage.Blob: + return storage_client.bucket(BUCKET_NAME).blob( + f"results/{file_type.value}/{id}/{file_name}" + ) + + +def generate_upload_urls( + input: UploadInput, + id: str, +) -> JobUploadURLs: + expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=15) + video_blob = get_upload_blob(input.video_file_name, FileType.VIDEO, id) + logo_blob = get_upload_blob(input.logo_file_name, FileType.IMAGE, id) + video_gcs_uri = video_blob.generate_signed_url( + expiration=expiry, + method="PUT", + ) + logo_gcs_uri = logo_blob.generate_signed_url( + expiration=expiry, + method="PUT", + ) + return JobUploadURLs( + video_file=SignedURL(uri=video_gcs_uri, expiry=expiry), + logo_file=SignedURL(uri=logo_gcs_uri, expiry=expiry), + ) From e247b5d59e3ae2b57aea2136cd398ec0d501c658 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Fri, 10 Oct 2025 21:14:23 +0200 Subject: [PATCH 04/19] back up --- .gitignore | 3 ++- api/Dockerfile | 21 +++++++++++++++++++++ api/README.md | 22 ++++++++++++++++++++++ api/input.py | 2 ++ api/main.py | 6 +++--- api/models.py | 1 + api/output.py | 17 +++++++++-------- api/requirements.txt | 6 ++++++ api/services/db.py | 7 +++++-- worker/README.md | 1 + 10 files changed, 72 insertions(+), 14 deletions(-) create mode 100644 api/Dockerfile create mode 100644 api/README.md create mode 100644 api/requirements.txt create mode 100644 worker/README.md diff --git a/.gitignore b/.gitignore index b6e861d..ee37881 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,5 @@ test/* .envrc adc.json .DS_Store -logo_detection.db \ No newline at end of file +logo_detection.db +*.pyc \ No newline at end of file diff --git a/api/Dockerfile b/api/Dockerfile new file mode 100644 index 0000000..7006ab4 --- /dev/null +++ b/api/Dockerfile @@ -0,0 +1,21 @@ +FROM python:3.13-slim +EXPOSE 8080 + +WORKDIR /app + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +ENV PYTHONUNBUFFERED=1 \ + # Keeps Python from generating .pyc files in the container + PYTHONDONTWRITEBYTECODE=1 \ + # Turns off buffering for easier container logging + UV_SYSTEM_PYTHON=1 + +COPY requirements.txt . + +RUN uv pip install --system --no-cache -r requirements.txt + +COPY . . + + +CMD ["uvicorn", "main:api", "--host", "0.0.0.0", "--port", "8080"] \ No newline at end of file diff --git a/api/README.md b/api/README.md new file mode 100644 index 0000000..47ee680 --- /dev/null +++ b/api/README.md @@ -0,0 +1,22 @@ +# Video Logo Tracker API + +FastAPI application for video logo processing with GCP integration. + +## Run API Locally with Docker + +1. **Download GCP credentials** and save as `adc.json` in `/api` directory. + +2. **Build and run**: + +```bash +cd api +docker build -t api . +docker run -v $(pwd)/adc.json:/app/credentials/gcp-key.json:ro \ + -e GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json \ + -p 8000:8080 api +``` + +3. **Access API**: + +- Docs: http://localhost:8000/docs +- Health: http://localhost:8000/health diff --git a/api/input.py b/api/input.py index 955b1aa..e246062 100644 --- a/api/input.py +++ b/api/input.py @@ -1,5 +1,7 @@ from pydantic import BaseModel + from typing import Optional + from models import JobStatus from services.db import JobDB diff --git a/api/main.py b/api/main.py index 17184c3..091684e 100644 --- a/api/main.py +++ b/api/main.py @@ -1,9 +1,9 @@ -from fastapi import FastAPI from contextlib import asynccontextmanager +from fastapi import FastAPI -from typing import List -from logging import getLogger, INFO from datetime import datetime, timezone +from logging import getLogger, INFO +from typing import List import uuid from errors import ( diff --git a/api/models.py b/api/models.py index e0c294c..92a01fe 100644 --- a/api/models.py +++ b/api/models.py @@ -1,4 +1,5 @@ from pydantic import BaseModel + from datetime import datetime from enum import Enum diff --git a/api/output.py b/api/output.py index c3e9019..b37bcfc 100644 --- a/api/output.py +++ b/api/output.py @@ -1,12 +1,13 @@ -from pydantic import BaseModel -from typing import Optional, List +from pydantic import BaseModel, computed_field + from datetime import datetime -from models import JobStatus -from db import JobDB +from logging import getLogger from pathlib import Path -from pydantic import computed_field +from typing import Optional, List from urllib.parse import urlparse -from logging import getLogger + +from models import JobStatus +from services.db import JobDB LOGGER = getLogger(__name__) @@ -25,9 +26,9 @@ class JobOutput(BaseModel): id: str video: GCSObject logo: GCSObject - output: Optional[GCSObject] + output: Optional[GCSObject] = None status: JobStatus - error: Optional[str] + error: Optional[str] = None created_at: datetime updated_at: datetime diff --git a/api/requirements.txt b/api/requirements.txt new file mode 100644 index 0000000..e9dafc2 --- /dev/null +++ b/api/requirements.txt @@ -0,0 +1,6 @@ +aiosqlite==0.21.0 +fastapi==0.118.1 +google-cloud-storage==3.4.0 +pydantic==2.12.0 +urllib3==2.5.0 +uvicorn==0.37.0 diff --git a/api/services/db.py b/api/services/db.py index fdbafef..4c9e96f 100644 --- a/api/services/db.py +++ b/api/services/db.py @@ -1,8 +1,11 @@ import aiosqlite +from pydantic import BaseModel + +from datetime import datetime, timezone + from models import JobStatus from typing import List, Optional -from datetime import datetime, timezone -from pydantic import BaseModel + DB_PATH = "logo_detection.db" diff --git a/worker/README.md b/worker/README.md new file mode 100644 index 0000000..496e81b --- /dev/null +++ b/worker/README.md @@ -0,0 +1 @@ +# video-logo-tracker worker From 6a761fc244b0c8f0f5267f9336984176b88ae151 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Fri, 10 Oct 2025 21:38:00 +0200 Subject: [PATCH 05/19] deploy api to cloud run --- .github/workflows/deploy.yml | 58 ++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) create mode 100644 .github/workflows/deploy.yml diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml new file mode 100644 index 0000000..36b083b --- /dev/null +++ b/.github/workflows/deploy.yml @@ -0,0 +1,58 @@ +name: Deploy API to Cloud Run + +on: + push: + branches: + - main + workflow_dispatch: + +env: + PROJECT_ID: jing-264314 + REGION: europe-west1 + API_SERVICE_NAME: video-logo-api + +jobs: + deploy-api: + runs-on: ubuntu-latest + + container: + image: google/cloud-sdk:latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Authenticate to Google Cloud + run: | + echo '${{ secrets.GCP_SA_KEY }}' > /tmp/gcp-key.json + gcloud auth activate-service-account --key-file=/tmp/gcp-key.json + gcloud config set project ${{ env.PROJECT_ID }} + + - name: Configure Docker for GCR + run: gcloud auth configure-docker + + - name: Build Docker image + working-directory: ./api + run: | + docker build -t gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} . + docker tag gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:latest + + - name: Push to Container Registry + run: | + docker push gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} + docker push gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:latest + + - name: Deploy to Cloud Run + run: | + gcloud run deploy ${{ env.API_SERVICE_NAME }} \ + --image gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + --region ${{ env.REGION }} \ + --allow-unauthenticated \ + --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ + --set-secrets=/app/credentials/gcp-key.json=neurons-cloud-function-service-account-key:latest \ + --set-env-vars GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json + + - name: Remove GCP key file + if: always() + run: rm -f /tmp/gcp-key.json From 1f73ac3a5288e07a80304febf25ee122eec4500a Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Fri, 10 Oct 2025 21:40:08 +0200 Subject: [PATCH 06/19] allow action on pr --- .github/workflows/deploy.yml | 46 ++++++++++++++++++++---------------- 1 file changed, 26 insertions(+), 20 deletions(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 36b083b..c497f7f 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -4,55 +4,61 @@ on: push: branches: - main + pull_request: + branches: + - main workflow_dispatch: env: PROJECT_ID: jing-264314 REGION: europe-west1 API_SERVICE_NAME: video-logo-api + ARTIFACT_REGISTRY: europe-west1-docker.pkg.dev jobs: deploy-api: runs-on: ubuntu-latest - container: - image: google/cloud-sdk:latest - steps: - name: Checkout code uses: actions/checkout@v4 - name: Authenticate to Google Cloud - run: | - echo '${{ secrets.GCP_SA_KEY }}' > /tmp/gcp-key.json - gcloud auth activate-service-account --key-file=/tmp/gcp-key.json - gcloud config set project ${{ env.PROJECT_ID }} + uses: google-github-actions/auth@v2 + with: + credentials_json: ${{ secrets.GCP_SA_KEY }} - - name: Configure Docker for GCR - run: gcloud auth configure-docker + - name: Set up Cloud SDK + uses: google-github-actions/setup-gcloud@v2 + + - name: Configure Docker for Artifact Registry + run: gcloud auth configure-docker ${{ env.ARTIFACT_REGISTRY }} - name: Build Docker image working-directory: ./api run: | - docker build -t gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} . - docker tag gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ - gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:latest + docker build -t ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} . + docker tag ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest - - name: Push to Container Registry + - name: Push to Artifact Registry run: | - docker push gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} - docker push gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:latest + docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} + docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest - name: Deploy to Cloud Run run: | gcloud run deploy ${{ env.API_SERVICE_NAME }} \ - --image gcr.io/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + --image ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ --region ${{ env.REGION }} \ --allow-unauthenticated \ --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ --set-secrets=/app/credentials/gcp-key.json=neurons-cloud-function-service-account-key:latest \ - --set-env-vars GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json + --set-env-vars GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json \ + --max-instances 2 - - name: Remove GCP key file - if: always() - run: rm -f /tmp/gcp-key.json + - name: Show deployment URL + run: | + echo "Service URL: $(gcloud run services describe ${{ env.API_SERVICE_NAME }} \ + --region ${{ env.REGION }} \ + --format 'value(status.url)')" From a5337d2089da8d01f92a6cc39944013212fbecd4 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 11:24:21 +0200 Subject: [PATCH 07/19] allow lovable app to send request to this API --- .github/workflows/deploy.yml | 98 ++++++++++++++++++------------------ api/main.py | 10 +++- 2 files changed, 58 insertions(+), 50 deletions(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index c497f7f..8cd5ca1 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -1,64 +1,64 @@ name: Deploy API to Cloud Run on: - push: - branches: - - main - pull_request: - branches: - - main - workflow_dispatch: + push: + branches: + - main + pull_request: + branches: + - main + workflow_dispatch: env: - PROJECT_ID: jing-264314 - REGION: europe-west1 - API_SERVICE_NAME: video-logo-api - ARTIFACT_REGISTRY: europe-west1-docker.pkg.dev + PROJECT_ID: jing-264314 + REGION: europe-west1 + API_SERVICE_NAME: video-logo-api + ARTIFACT_REGISTRY: europe-west1-docker.pkg.dev jobs: - deploy-api: - runs-on: ubuntu-latest + deploy-api: + runs-on: ubuntu-latest - steps: - - name: Checkout code - uses: actions/checkout@v4 + steps: + - name: Checkout code + uses: actions/checkout@v4 - - name: Authenticate to Google Cloud - uses: google-github-actions/auth@v2 - with: - credentials_json: ${{ secrets.GCP_SA_KEY }} + - name: Authenticate to Google Cloud + uses: google-github-actions/auth@v2 + with: + credentials_json: ${{ secrets.GCP_SA_KEY }} - - name: Set up Cloud SDK - uses: google-github-actions/setup-gcloud@v2 + - name: Set up Cloud SDK + uses: google-github-actions/setup-gcloud@v2 - - name: Configure Docker for Artifact Registry - run: gcloud auth configure-docker ${{ env.ARTIFACT_REGISTRY }} + - name: Configure Docker for Artifact Registry + run: gcloud auth configure-docker ${{ env.ARTIFACT_REGISTRY }} - - name: Build Docker image - working-directory: ./api - run: | - docker build -t ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} . - docker tag ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ - ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest + - name: Build Docker image + working-directory: ./api + run: | + docker build -t ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} . + docker tag ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest - - name: Push to Artifact Registry - run: | - docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} - docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest + - name: Push to Artifact Registry + run: | + docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} + docker push ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:latest - - name: Deploy to Cloud Run - run: | - gcloud run deploy ${{ env.API_SERVICE_NAME }} \ - --image ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ - --region ${{ env.REGION }} \ - --allow-unauthenticated \ - --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ - --set-secrets=/app/credentials/gcp-key.json=neurons-cloud-function-service-account-key:latest \ - --set-env-vars GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json \ - --max-instances 2 + - name: Deploy to Cloud Run + run: | + gcloud run deploy ${{ env.API_SERVICE_NAME }} \ + --image ${{ env.ARTIFACT_REGISTRY }}/${{ env.PROJECT_ID }}/${{ env.API_SERVICE_NAME }}/${{ env.API_SERVICE_NAME }}:${{ github.sha }} \ + --region ${{ env.REGION }} \ + --allow-unauthenticated \ + --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ + --set-secrets=/app/credentials/gcp-key.json=neurons-cloud-function-service-account-key:latest \ + --set-env-vars GOOGLE_APPLICATION_CREDENTIALS=/app/credentials/gcp-key.json \ + --max-instances 1 - - name: Show deployment URL - run: | - echo "Service URL: $(gcloud run services describe ${{ env.API_SERVICE_NAME }} \ - --region ${{ env.REGION }} \ - --format 'value(status.url)')" + - name: Show deployment URL + run: | + echo "Service URL: $(gcloud run services describe ${{ env.API_SERVICE_NAME }} \ + --region ${{ env.REGION }} \ + --format 'value(status.url)')" diff --git a/api/main.py b/api/main.py index 091684e..480c7b7 100644 --- a/api/main.py +++ b/api/main.py @@ -1,5 +1,6 @@ from contextlib import asynccontextmanager from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware from datetime import datetime, timezone from logging import getLogger, INFO @@ -34,7 +35,14 @@ async def lifespan(app: FastAPI): api = FastAPI(lifespan=lifespan) - +api.add_middleware( + CORSMiddleware, + allow_origins=[ + "https://8a33b7dc-2d6d-43a9-af02-5ff81682d7f1.lovableproject.com", + ], + allow_methods=["*"], + allow_headers=["*"], +) @api.get("/health") async def health(): From b70c3c543de32a8bbac0145f624ccad521bf5103 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 11:48:37 +0200 Subject: [PATCH 08/19] add content type in signed url --- api/services/storage.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/api/services/storage.py b/api/services/storage.py index b9dbcb6..028c8e8 100644 --- a/api/services/storage.py +++ b/api/services/storage.py @@ -34,10 +34,12 @@ def generate_upload_urls( video_gcs_uri = video_blob.generate_signed_url( expiration=expiry, method="PUT", + content_type="video/mp4", ) logo_gcs_uri = logo_blob.generate_signed_url( expiration=expiry, method="PUT", + content_type="image/jpeg", ) return JobUploadURLs( video_file=SignedURL(uri=video_gcs_uri, expiry=expiry), From 6cf96deba7624d5fae9a443c307083c69db324f6 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 12:29:17 +0200 Subject: [PATCH 09/19] update readme --- api/README.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/api/README.md b/api/README.md index 47ee680..fb02ec3 100644 --- a/api/README.md +++ b/api/README.md @@ -18,5 +18,4 @@ docker run -v $(pwd)/adc.json:/app/credentials/gcp-key.json:ro \ 3. **Access API**: -- Docs: http://localhost:8000/docs -- Health: http://localhost:8000/health +- Docs: https://video-logo-api-606507129150.europe-west1.run.app/docs# From 3a88e086cab22a47d9f06999c17d355c18cc37fa Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 14:57:58 +0200 Subject: [PATCH 10/19] back up --- api/README.md | 3 +- api/errors.py | 8 +++ api/input.py | 13 ++--- api/main.py | 24 ++++++--- api/models.py | 17 +++++- api/output.py | 43 ++++++++------- api/services/db.py | 56 +++++++++---------- api/services/storage.py | 70 +++++++++++++++--------- worker/main.py | 116 +++++++++++++++++++++++----------------- worker/models.py | 25 +++++++++ worker/requirements.txt | 8 +++ worker/sift.py | 18 +++---- worker/storage.py | 36 +++++++++++++ 13 files changed, 291 insertions(+), 146 deletions(-) create mode 100644 worker/models.py create mode 100644 worker/requirements.txt create mode 100644 worker/storage.py diff --git a/api/README.md b/api/README.md index fb02ec3..c6c599f 100644 --- a/api/README.md +++ b/api/README.md @@ -18,4 +18,5 @@ docker run -v $(pwd)/adc.json:/app/credentials/gcp-key.json:ro \ 3. **Access API**: -- Docs: https://video-logo-api-606507129150.europe-west1.run.app/docs# +-Production Docs: https://video-logo-api-606507129150.europe-west1.run.app/docs# +-Local Docs: http://localhost:8000/docs# diff --git a/api/errors.py b/api/errors.py index 3b428ad..386ebc7 100644 --- a/api/errors.py +++ b/api/errors.py @@ -39,3 +39,11 @@ def __init__(self, job_id: str): status_code=404, detail=f"Job not found with id {job_id}", ) + + +class InvalidGCSURLException(HTTPException): + def __init__(self, url: str): + super().__init__( + status_code=400, + detail=f"Invalid GCS URL: {url}", + ) diff --git a/api/input.py b/api/input.py index e246062..b6248a8 100644 --- a/api/input.py +++ b/api/input.py @@ -12,17 +12,18 @@ class UploadInput(BaseModel): class JobPatchInput(BaseModel): - output_gcs_uri: Optional[str] + output_gcs_url: Optional[str] = None status: JobStatus - error: Optional[str] + error: Optional[str] = None def to_job_db(self, job: JobDB) -> JobDB: job.status = self.status - if self.output_gcs_uri: - job.output_gcs_uri = self.output_gcs_uri + if self.output_gcs_url: + job.output_gcs_url = self.output_gcs_url if self.error: job.error = self.error - if self.status: - job.status = self.status return job + + def is_gcs_url(self, url: str) -> bool: + return url.startswith("gs://") diff --git a/api/main.py b/api/main.py index 480c7b7..8246d40 100644 --- a/api/main.py +++ b/api/main.py @@ -13,13 +13,14 @@ JobNotFoundException, MissingVideoFileException, MissingLogoFileException, + InvalidGCSURLException, ) from models import JobStatus from input import UploadInput, JobPatchInput from output import JobOutput from services.db import get, save, delete, list, init_db, drop_db, JobDB -from services.storage import generate_upload_urls +from services.storage import generate_upload_urls, delete_by_url LOGGER = getLogger(__name__) LOGGER.setLevel(INFO) @@ -39,11 +40,13 @@ async def lifespan(app: FastAPI): CORSMiddleware, allow_origins=[ "https://8a33b7dc-2d6d-43a9-af02-5ff81682d7f1.lovableproject.com", + "https://id-preview--8a33b7dc-2d6d-43a9-af02-5ff81682d7f1.lovable.app", ], allow_methods=["*"], allow_headers=["*"], ) + @api.get("/health") async def health(): return {"status": "ok"} @@ -53,11 +56,11 @@ async def health(): async def create_job(input: UploadInput) -> JobOutput: validate_upload_input(input) id = str(uuid.uuid4()) - upload_urls = generate_upload_urls(input, id) + upload_urls = generate_upload_urls(input.video_file_name, input.logo_file_name, id) job = JobDB( id=id, - video_gcs_uri=upload_urls.video_file.uri, - logo_gcs_uri=upload_urls.logo_file.uri, + video_gcs_url=upload_urls.video_file.url, + logo_gcs_url=upload_urls.logo_file.url, status=JobStatus.UPLOADING, created_at=datetime.now(tz=timezone.utc), ) @@ -83,7 +86,7 @@ async def run_job(id: str) -> JobOutput: raise JobNotFoundException(id) job.status = JobStatus.RUNNING job.error = None - job.output_gcs_uri = None + job.output_gcs_url = None await save(job) return JobOutput.from_job_db(job) @@ -95,6 +98,8 @@ async def list_jobs() -> List[JobOutput]: @api.patch("/jobs/{id}", response_model=JobOutput) async def job_done(id: str, input: JobPatchInput) -> JobOutput: + if input.output_gcs_url and not input.is_gcs_url(input.output_gcs_url): + raise InvalidGCSURLException(input.output_gcs_url) job = await get(id) if job is None: raise JobNotFoundException(id) @@ -105,7 +110,13 @@ async def job_done(id: str, input: JobPatchInput) -> JobOutput: @api.delete("/jobs/{id}", status_code=204) async def delete_job(id: str): - await delete(id) + job = await get(id) + if job: + await delete(id) + delete_by_url(job.video_gcs_url) + delete_by_url(job.logo_gcs_url) + if job.output_gcs_url: + delete_by_url(job.output_gcs_url) return @@ -116,7 +127,6 @@ def validate_upload_input(input: UploadInput): raise MissingVideoFileException() if logo_file_name is None or logo_file_name == "": raise MissingLogoFileException() - if not video_file_name.lower().endswith((".mp4")): raise InvalidVideoFileExtensionException(video_file_name) if not logo_file_name.lower().endswith((".jpg")): diff --git a/api/models.py b/api/models.py index 92a01fe..969a16f 100644 --- a/api/models.py +++ b/api/models.py @@ -1,7 +1,8 @@ -from pydantic import BaseModel +from pydantic import BaseModel, computed_field from datetime import datetime from enum import Enum +from pathlib import Path class FileType(str, Enum): @@ -9,9 +10,21 @@ class FileType(str, Enum): IMAGE = "image" +class SignedURLMethod(str, Enum): + GET = "GET" + PUT = "PUT" + + class SignedURL(BaseModel): - uri: str + url: str + signed_url: str expiry: datetime + method: SignedURLMethod + + @computed_field + @property + def name(self) -> str: + return Path(self.url).name class JobUploadURLs(BaseModel): diff --git a/api/output.py b/api/output.py index b37bcfc..c731097 100644 --- a/api/output.py +++ b/api/output.py @@ -1,32 +1,21 @@ -from pydantic import BaseModel, computed_field +from pydantic import BaseModel from datetime import datetime from logging import getLogger -from pathlib import Path from typing import Optional, List -from urllib.parse import urlparse -from models import JobStatus +from models import JobStatus, SignedURL, SignedURLMethod from services.db import JobDB +from services.storage import get_signed_url, blob_from_gcs_url LOGGER = getLogger(__name__) -class GCSObject(BaseModel): - uri: str - - @computed_field - @property - def name(self) -> str: - parsed = urlparse(self.uri) - return Path(parsed.path).name - - class JobOutput(BaseModel): id: str - video: GCSObject - logo: GCSObject - output: Optional[GCSObject] = None + video: SignedURL + logo: SignedURL + output: Optional[SignedURL] = None status: JobStatus error: Optional[str] = None created_at: datetime @@ -35,10 +24,24 @@ class JobOutput(BaseModel): def from_job_db(job_db: JobDB) -> "JobOutput": return JobOutput( id=job_db.id, - video=GCSObject(uri=job_db.video_gcs_uri), - logo=GCSObject(uri=job_db.logo_gcs_uri), + video=get_signed_url( + blob_from_gcs_url(job_db.video_gcs_url), + "video/mp4", + SignedURLMethod.GET, + ), + logo=get_signed_url( + blob_from_gcs_url(job_db.logo_gcs_url), + "image/jpeg", + SignedURLMethod.GET, + ), output=( - GCSObject(uri=job_db.output_gcs_uri) if job_db.output_gcs_uri else None + get_signed_url( + blob_from_gcs_url(job_db.output_gcs_url), + "video/mp4", + SignedURLMethod.GET, + ) + if job_db.output_gcs_url + else None ), status=job_db.status, error=job_db.error, diff --git a/api/services/db.py b/api/services/db.py index 4c9e96f..5253ee4 100644 --- a/api/services/db.py +++ b/api/services/db.py @@ -12,9 +12,9 @@ class JobDB(BaseModel): id: str - video_gcs_uri: str - logo_gcs_uri: str - output_gcs_uri: Optional[str] = None + video_gcs_url: str + logo_gcs_url: str + output_gcs_url: Optional[str] = None status: JobStatus error: Optional[str] = None created_at: Optional[datetime] = None @@ -26,9 +26,9 @@ async def init_db(): query = f""" CREATE TABLE IF NOT EXISTS logo_detection ( id TEXT PRIMARY KEY, - video_gcs_uri TEXT NOT NULL, - logo_gcs_uri TEXT NOT NULL, - output_gcs_uri TEXT, + video_gcs_url TEXT NOT NULL, + logo_gcs_url TEXT NOT NULL, + output_gcs_url TEXT, status TEXT CHECK(status IN ({statuses})), error TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, @@ -52,9 +52,9 @@ async def get(id: str) -> JobDB: """ SELECT id, - video_gcs_uri, - logo_gcs_uri, - output_gcs_uri, + video_gcs_url, + logo_gcs_url, + output_gcs_url, status, error, created_at, @@ -71,9 +71,9 @@ async def get(id: str) -> JobDB: return None return JobDB( id=row[0], - video_gcs_uri=row[1], - logo_gcs_uri=row[2], - output_gcs_uri=row[3], + video_gcs_url=row[1], + logo_gcs_url=row[2], + output_gcs_url=row[3], status=row[4], error=row[5], created_at=row[6], @@ -87,22 +87,22 @@ async def save(job: JobDB): await db.execute( """INSERT INTO logo_detection (id, - video_gcs_uri, - logo_gcs_uri, - output_gcs_uri, + video_gcs_url, + logo_gcs_url, + output_gcs_url, status, error, created_at, updated_at) VALUES - (:id, :video_gcs_uri, :logo_gcs_uri, :output_gcs_uri, :status, :error, :created_at, :updated_at) + (:id, :video_gcs_url, :logo_gcs_url, :output_gcs_url, :status, :error, :created_at, :updated_at) ON CONFLICT(id) DO UPDATE SET - video_gcs_uri=:video_gcs_uri, - logo_gcs_uri=:logo_gcs_uri, - output_gcs_uri=:output_gcs_uri, + video_gcs_url=:video_gcs_url, + logo_gcs_url=:logo_gcs_url, + output_gcs_url=:output_gcs_url, status=:status, error=:error, updated_at=:updated_at @@ -110,9 +110,9 @@ async def save(job: JobDB): """, { "id": job.id, - "video_gcs_uri": job.video_gcs_uri, - "logo_gcs_uri": job.logo_gcs_uri, - "output_gcs_uri": job.output_gcs_uri, + "video_gcs_url": job.video_gcs_url, + "logo_gcs_url": job.logo_gcs_url, + "output_gcs_url": job.output_gcs_url, "status": job.status, "error": job.error, "created_at": job.created_at, @@ -134,9 +134,9 @@ async def list() -> List[JobDB]: """ SELECT id, - video_gcs_uri, - logo_gcs_uri, - output_gcs_uri, + video_gcs_url, + logo_gcs_url, + output_gcs_url, status, error, created_at, @@ -149,9 +149,9 @@ async def list() -> List[JobDB]: return [ JobDB( id=row[0], - video_gcs_uri=row[1], - logo_gcs_uri=row[2], - output_gcs_uri=row[3], + video_gcs_url=row[1], + logo_gcs_url=row[2], + output_gcs_url=row[3], status=row[4], error=row[5], created_at=row[6], diff --git a/api/services/storage.py b/api/services/storage.py index 028c8e8..603345b 100644 --- a/api/services/storage.py +++ b/api/services/storage.py @@ -1,47 +1,69 @@ from google.cloud import storage +from google.api_core.exceptions import NotFound from datetime import datetime, timedelta, timezone -from input import UploadInput from models import ( FileType, SignedURL, + SignedURLMethod, JobUploadURLs, ) - +storage_client = storage.Client(project="jing-264314") BUCKET_NAME = "code_challenge_logo_detection" -storage_client = storage.Client(project="jing") +EXPIRY_MINUTES = 60 -def get_upload_blob(file_name: str, file_type: FileType, id: str) -> storage.Blob: - return storage_client.bucket(BUCKET_NAME).blob( - f"uploads/{file_type.value}/{id}/{file_name}" +def get_signed_url( + blob: storage.Blob, content_type: str, method: SignedURLMethod +) -> SignedURL: + expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=EXPIRY_MINUTES) + return SignedURL( + url=f"gs://{blob.bucket.name}/{blob.name}", + signed_url=blob.generate_signed_url( + expiration=expiry, + method=method, + content_type=content_type, + ), + expiry=expiry, + method=method, ) -def get_result_blob(file_name: str, file_type: FileType, id: str) -> storage.Blob: +def blob_from_gcs_url(gcs_url: str) -> storage.Blob: + if gcs_url.startswith("gs://"): + parts = gcs_url.replace("gs://", "").split("/", 1) + bucket_name, blob_path = parts[0], parts[1] + else: + raise ValueError(f"Invalid GCS URL: {gcs_url}") + return storage_client.bucket(bucket_name).blob(blob_path) + + +def blob_from_file_name(file_name: str, file_type: FileType, id: str) -> storage.Blob: return storage_client.bucket(BUCKET_NAME).blob( - f"results/{file_type.value}/{id}/{file_name}" + f"uploads/{id}/{file_type.value}/{file_name}" ) def generate_upload_urls( - input: UploadInput, + video_file_name: str, + logo_file_name: str, id: str, ) -> JobUploadURLs: - expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=15) - video_blob = get_upload_blob(input.video_file_name, FileType.VIDEO, id) - logo_blob = get_upload_blob(input.logo_file_name, FileType.IMAGE, id) - video_gcs_uri = video_blob.generate_signed_url( - expiration=expiry, - method="PUT", - content_type="video/mp4", - ) - logo_gcs_uri = logo_blob.generate_signed_url( - expiration=expiry, - method="PUT", - content_type="image/jpeg", - ) + video_blob = blob_from_file_name(video_file_name, FileType.VIDEO, id) + video_url = get_signed_url(video_blob, "video/mp4", SignedURLMethod.PUT) + logo_blob = blob_from_file_name(logo_file_name, FileType.IMAGE, id) + logo_url = get_signed_url(logo_blob, "image/jpeg", SignedURLMethod.PUT) + return JobUploadURLs( - video_file=SignedURL(uri=video_gcs_uri, expiry=expiry), - logo_file=SignedURL(uri=logo_gcs_uri, expiry=expiry), + video_file=video_url, + logo_file=logo_url, ) + + +def delete_by_url(url: str): + try: + blob = blob_from_gcs_url(url) + blob.delete() + except NotFound: + return + return diff --git a/worker/main.py b/worker/main.py index 98a5824..a1b91d2 100644 --- a/worker/main.py +++ b/worker/main.py @@ -1,20 +1,22 @@ -from typing import Dict - import functions_framework from flask import Request from google.cloud import storage from pydantic import BaseModel, ValidationError - +import requests +from typing import Dict +from pathlib import Path from sift import detect_logo_with_sift +from models import JobStatus, JobResult +from storage import download_from_gcs, upload_to_gcs BUCKET_NAME = "code_challenge_logo_detection" storage_client = storage.Client(project="jing") class ProcessVideoRequest(BaseModel): + id: str video_gcs_path: str logo_gcs_path: str - output_gcs_path: str @functions_framework.http @@ -26,54 +28,70 @@ def process_video(request: Request): except ValidationError as e: return {"error": "Invalid request", "details": e.errors()}, 400 - video_path = request.video_gcs_path - logo_path = request.logo_gcs_path - output_path = request.output_gcs_path - - bucket = storage_client.bucket(BUCKET_NAME) - - local_video_path = get_local_path(video_path) - local_logo_path = get_local_path(logo_path) - local_output_path = get_local_path(output_path) - - video_blob = bucket.blob(video_path) - video_blob.download_to_filename(local_video_path) - - logo_blob = bucket.blob(logo_path) - logo_blob.download_to_filename(local_logo_path) - - detect_logo_with_sift( - logo_path=local_logo_path, - video_path=local_video_path, - output_path=local_output_path, - min_matches=15, - ratio_threshold=0.75, + result = JobResult( + id=request.id, + video_gcs_path=request.video_gcs_path, + logo_gcs_path=request.logo_gcs_path, + status=JobStatus.RUNNING, ) + local_output_path = get_local_path(request.video_gcs_path, request.id) - output_blob = bucket.blob(output_path) - output_blob.upload_from_filename(output_path) - - return {"result": output_blob.path} - - -def get_local_path(path: str) -> str: - return f"/tmp/{path}" + try: + detect_logo_with_sift( + logo_path=download_from_gcs(request.logo_gcs_path, request.id), + video_path=download_from_gcs(request.video_gcs_path, request.id), + output_path=local_output_path, + min_matches=15, + ratio_threshold=0.75, + ) + except Exception as e: + error = str(e) + result.status = JobStatus.FAILED + result.error = error + update_job_status(result) + return result + + blob_name = upload_to_gcs(local_output_path, request.id) + result.output_gcs_path = blob_name + result.status = JobStatus.SUCCESS + update_job_status(result) + + return result + + +def get_local_path(video_gcs_path: str, id: str) -> str: + file_name = Path(video_gcs_path).name + return f"/tmp/{file_name}" + + +def update_job_status(result: JobResult): + resp = requests.patch( + f"https://video-logo-api-606507129150.europe-west1.run.app/jobs/{result.id}", + json={ + "output_gcs_url": result.output_gcs_path, + "status": result.status, + "error": result.error, + }, + ) + if resp.status_code != 200: + raise Exception(f"Failed to update job status: {resp.status_code} {resp.text}") + return -# if __name__ == "__main__": -# from werkzeug.test import EnvironBuilder +if __name__ == "__main__": + from werkzeug.test import EnvironBuilder -# builder = EnvironBuilder( -# method="POST", -# path="/process", -# json={ -# "video_gcs_path": "Video_1.mp4", -# "logo_gcs_path": "neurons_logo.jpg", -# "output_gcs_path": "output_video_1_fearure_matching_SIFT.mp4", -# }, -# ) -# env = builder.get_environ() -# req = Request(env) -# response = process_video(req) -# print(response) + builder = EnvironBuilder( + method="POST", + path="/process", + json={ + "id": "72956fa6-2415-49cc-8108-e068e4a3075d", + "logo_gcs_path": "gs://code_challenge_logo_detection/uploads/72956fa6-2415-49cc-8108-e068e4a3075d/image/neurons_logo.jpg", + "video_gcs_path": "gs://code_challenge_logo_detection/uploads/72956fa6-2415-49cc-8108-e068e4a3075d/video/Video_1.mp4", + }, + ) + env = builder.get_environ() + req = Request(env) +response = process_video(req) +print(response) diff --git a/worker/models.py b/worker/models.py new file mode 100644 index 0000000..a2508a8 --- /dev/null +++ b/worker/models.py @@ -0,0 +1,25 @@ +from pydantic import BaseModel +from typing import Optional + +from enum import Enum + + +class JobStatus(str, Enum): + RUNNING = "running" + SUCCESS = "success" + FAILED = "failed" + + +class UpdateJobStatusRequest(BaseModel): + output_gcs_url: Optional[str] = None + status: JobStatus + error: Optional[str] = None + + +class JobResult(BaseModel): + id: str + video_gcs_path: str + logo_gcs_path: str + output_gcs_path: Optional[str] = None + status: Optional[JobStatus] = None + error: Optional[str] = None diff --git a/worker/requirements.txt b/worker/requirements.txt new file mode 100644 index 0000000..fa1aa91 --- /dev/null +++ b/worker/requirements.txt @@ -0,0 +1,8 @@ +functions-framework +flask +google-cloud-storage==3.4.0 +pydantic==2.12.0 +requests +opencv-python-headless +numpy + diff --git a/worker/sift.py b/worker/sift.py index d9165d6..3271671 100644 --- a/worker/sift.py +++ b/worker/sift.py @@ -119,14 +119,14 @@ def get_video_info(cap: cv2.VideoCapture) -> tuple[int, int, int, int]: return fps, width, height, total_frames -if __name__ == "__main__": +# if __name__ == "__main__": - logo_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/neurons_logo.jpg" - video_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/Video_1.mp4" - output_path = "/Users/jing/code_challenge/video-logo-tracker/output_video_1_fearure_matching_SIFT.mp4" +# logo_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/neurons_logo.jpg" +# video_path = "/Users/jing/code_challenge/video-logo-tracker/senior_ai_engineer_assignment/Video_1.mp4" +# output_path = "/Users/jing/code_challenge/video-logo-tracker/output_video_1_fearure_matching_SIFT.mp4" - detect_logo_with_sift( - logo_path=logo_path, - video_path=video_path, - output_path=output_path, - ) +# detect_logo_with_sift( +# logo_path=logo_path, +# video_path=video_path, +# output_path=output_path, +# ) diff --git a/worker/storage.py b/worker/storage.py new file mode 100644 index 0000000..39424e1 --- /dev/null +++ b/worker/storage.py @@ -0,0 +1,36 @@ +from google.cloud import storage + +from pathlib import Path + +storage_client = storage.Client(project="jing-264314") + + +def download_from_gcs(gcs_url, id: str): + local_dir = Path(f"/tmp/{id}/input") + local_dir.mkdir(parents=True, exist_ok=True) + local_path = local_dir / Path(gcs_url).name + + # Parse gs:// URL properly + if gcs_url.startswith("gs://"): + parts = gcs_url.replace("gs://", "").split("/", 1) + bucket_name, blob_path = parts[0], parts[1] + else: + raise ValueError(f"Invalid GCS URl: {gcs_url}") + + bucket = storage_client.bucket(bucket_name) + blob = bucket.blob(blob_path) + print(f"Downloading {gcs_url} to {local_path}") + blob.download_to_filename(local_path) + print(f"Downloaded {gcs_url} to {local_path}") + return local_path + + +def upload_to_gcs(local_path: str, id: str) -> str: + file_name = Path(local_path).name + blob = storage_client.bucket("code_challenge_logo_detection").blob( + f"results/{id}/{file_name}" + ) + blob.upload_from_filename(local_path) + gcs_url = f"gs://{blob.bucket.name}/{blob.name}" + print(f"Result uploaded to {gcs_url}") + return gcs_url From 409be8acf9618c9dec9965041bd95bdc2cb49dbf Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 15:43:15 +0200 Subject: [PATCH 11/19] fix signurl method for upload --- api/main.py | 24 +++++++++++++++--------- api/models.py | 5 ----- api/output.py | 27 +++++++++++++++++++++++++-- api/services/storage.py | 23 +++++------------------ 4 files changed, 45 insertions(+), 34 deletions(-) diff --git a/api/main.py b/api/main.py index 8246d40..ccc0999 100644 --- a/api/main.py +++ b/api/main.py @@ -17,10 +17,10 @@ ) from models import JobStatus from input import UploadInput, JobPatchInput -from output import JobOutput +from output import JobOutput, CreateJobOutput from services.db import get, save, delete, list, init_db, drop_db, JobDB -from services.storage import generate_upload_urls, delete_by_url +from services.storage import delete_by_url, get_gcs_url, blob_from_file_name, FileType LOGGER = getLogger(__name__) LOGGER.setLevel(INFO) @@ -52,21 +52,27 @@ async def health(): return {"status": "ok"} -@api.post("/jobs", response_model=JobOutput) -async def create_job(input: UploadInput) -> JobOutput: +@api.post("/jobs", response_model=CreateJobOutput) +async def create_job(input: UploadInput) -> CreateJobOutput: validate_upload_input(input) id = str(uuid.uuid4()) - upload_urls = generate_upload_urls(input.video_file_name, input.logo_file_name, id) job = JobDB( id=id, - video_gcs_url=upload_urls.video_file.url, - logo_gcs_url=upload_urls.logo_file.url, + video_gcs_url=get_gcs_url( + blob_from_file_name(input.video_file_name, FileType.VIDEO, id) + ), + logo_gcs_url=get_gcs_url( + blob_from_file_name(input.logo_file_name, FileType.IMAGE, id) + ), status=JobStatus.UPLOADING, created_at=datetime.now(tz=timezone.utc), ) await save(job) - saved_job = await get(id) - return JobOutput.from_job_db(saved_job) + return CreateJobOutput.from_files( + input.video_file_name, + input.logo_file_name, + id, + ) @api.get("/jobs/{id}", response_model=JobOutput) diff --git a/api/models.py b/api/models.py index 969a16f..d8c94dc 100644 --- a/api/models.py +++ b/api/models.py @@ -27,11 +27,6 @@ def name(self) -> str: return Path(self.url).name -class JobUploadURLs(BaseModel): - video_file: SignedURL - logo_file: SignedURL - - class JobStatus(str, Enum): UPLOADING = "uploading" RUNNING = "running" diff --git a/api/output.py b/api/output.py index c731097..0c8fb99 100644 --- a/api/output.py +++ b/api/output.py @@ -4,13 +4,36 @@ from logging import getLogger from typing import Optional, List -from models import JobStatus, SignedURL, SignedURLMethod +from models import JobStatus, SignedURL, SignedURLMethod, FileType from services.db import JobDB -from services.storage import get_signed_url, blob_from_gcs_url +from services.storage import blob_from_file_name, get_signed_url, blob_from_gcs_url LOGGER = getLogger(__name__) +class CreateJobOutput(BaseModel): + id: str + video: SignedURL + logo: SignedURL + + def from_files( + video_file_name: str, logo_file_name: str, id: str + ) -> "CreateJobOutput": + return CreateJobOutput( + id=id, + video=get_signed_url( + blob_from_file_name(video_file_name, FileType.VIDEO, id), + "video/mp4", + SignedURLMethod.PUT, + ), + logo=get_signed_url( + blob_from_file_name(logo_file_name, FileType.IMAGE, id), + "image/jpeg", + SignedURLMethod.PUT, + ), + ) + + class JobOutput(BaseModel): id: str video: SignedURL diff --git a/api/services/storage.py b/api/services/storage.py index 603345b..a5cc4a8 100644 --- a/api/services/storage.py +++ b/api/services/storage.py @@ -5,7 +5,6 @@ FileType, SignedURL, SignedURLMethod, - JobUploadURLs, ) storage_client = storage.Client(project="jing-264314") @@ -18,7 +17,7 @@ def get_signed_url( ) -> SignedURL: expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=EXPIRY_MINUTES) return SignedURL( - url=f"gs://{blob.bucket.name}/{blob.name}", + url=get_gcs_url(blob), signed_url=blob.generate_signed_url( expiration=expiry, method=method, @@ -29,6 +28,10 @@ def get_signed_url( ) +def get_gcs_url(blob: storage.Blob) -> str: + return f"gs://{blob.bucket.name}/{blob.name}" + + def blob_from_gcs_url(gcs_url: str) -> storage.Blob: if gcs_url.startswith("gs://"): parts = gcs_url.replace("gs://", "").split("/", 1) @@ -44,22 +47,6 @@ def blob_from_file_name(file_name: str, file_type: FileType, id: str) -> storage ) -def generate_upload_urls( - video_file_name: str, - logo_file_name: str, - id: str, -) -> JobUploadURLs: - video_blob = blob_from_file_name(video_file_name, FileType.VIDEO, id) - video_url = get_signed_url(video_blob, "video/mp4", SignedURLMethod.PUT) - logo_blob = blob_from_file_name(logo_file_name, FileType.IMAGE, id) - logo_url = get_signed_url(logo_blob, "image/jpeg", SignedURLMethod.PUT) - - return JobUploadURLs( - video_file=video_url, - logo_file=logo_url, - ) - - def delete_by_url(url: str): try: blob = blob_from_gcs_url(url) From 0bc877bcba52509167c03ca4522166958dc8c61a Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 15:55:39 +0200 Subject: [PATCH 12/19] don't include content type for get signed url --- api/output.py | 27 ++++++++++++--------------- api/services/storage.py | 6 +++++- worker/main.py | 6 +++--- 3 files changed, 20 insertions(+), 19 deletions(-) diff --git a/api/output.py b/api/output.py index 0c8fb99..371c58e 100644 --- a/api/output.py +++ b/api/output.py @@ -22,14 +22,14 @@ def from_files( return CreateJobOutput( id=id, video=get_signed_url( - blob_from_file_name(video_file_name, FileType.VIDEO, id), - "video/mp4", - SignedURLMethod.PUT, + blob=blob_from_file_name(video_file_name, FileType.VIDEO, id), + method=SignedURLMethod.PUT, + content_type="video/mp4", ), logo=get_signed_url( - blob_from_file_name(logo_file_name, FileType.IMAGE, id), - "image/jpeg", - SignedURLMethod.PUT, + blob=blob_from_file_name(logo_file_name, FileType.IMAGE, id), + method=SignedURLMethod.PUT, + content_type="image/jpeg", ), ) @@ -48,20 +48,17 @@ def from_job_db(job_db: JobDB) -> "JobOutput": return JobOutput( id=job_db.id, video=get_signed_url( - blob_from_gcs_url(job_db.video_gcs_url), - "video/mp4", - SignedURLMethod.GET, + blob=blob_from_gcs_url(job_db.video_gcs_url), + method=SignedURLMethod.GET, ), logo=get_signed_url( - blob_from_gcs_url(job_db.logo_gcs_url), - "image/jpeg", - SignedURLMethod.GET, + blob=blob_from_gcs_url(job_db.logo_gcs_url), + method=SignedURLMethod.GET, ), output=( get_signed_url( - blob_from_gcs_url(job_db.output_gcs_url), - "video/mp4", - SignedURLMethod.GET, + blob=blob_from_gcs_url(job_db.output_gcs_url), + method=SignedURLMethod.GET, ) if job_db.output_gcs_url else None diff --git a/api/services/storage.py b/api/services/storage.py index a5cc4a8..4a7aa99 100644 --- a/api/services/storage.py +++ b/api/services/storage.py @@ -1,19 +1,23 @@ from google.cloud import storage from google.api_core.exceptions import NotFound + from datetime import datetime, timedelta, timezone +from typing import Optional + from models import ( FileType, SignedURL, SignedURLMethod, ) + storage_client = storage.Client(project="jing-264314") BUCKET_NAME = "code_challenge_logo_detection" EXPIRY_MINUTES = 60 def get_signed_url( - blob: storage.Blob, content_type: str, method: SignedURLMethod + blob: storage.Blob, method: SignedURLMethod, content_type: Optional[str] = None ) -> SignedURL: expiry = datetime.now(tz=timezone.utc) + timedelta(minutes=EXPIRY_MINUTES) return SignedURL( diff --git a/worker/main.py b/worker/main.py index a1b91d2..8ce70d2 100644 --- a/worker/main.py +++ b/worker/main.py @@ -86,9 +86,9 @@ def update_job_status(result: JobResult): method="POST", path="/process", json={ - "id": "72956fa6-2415-49cc-8108-e068e4a3075d", - "logo_gcs_path": "gs://code_challenge_logo_detection/uploads/72956fa6-2415-49cc-8108-e068e4a3075d/image/neurons_logo.jpg", - "video_gcs_path": "gs://code_challenge_logo_detection/uploads/72956fa6-2415-49cc-8108-e068e4a3075d/video/Video_1.mp4", + "id": "df9c2228-143e-452f-a7be-744a85fad351", + "logo_gcs_path": "gs://code_challenge_logo_detection/uploads/df9c2228-143e-452f-a7be-744a85fad351/image/neurons_logo.jpg", + "video_gcs_path": "gs://code_challenge_logo_detection/uploads/df9c2228-143e-452f-a7be-744a85fad351/video/Video_1.mp4", }, ) env = builder.get_environ() From 4dce462987de7d263304215fca5569bdea763fef Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 16:19:28 +0200 Subject: [PATCH 13/19] download video instead of streaming --- api/services/storage.py | 1 + worker/main.py | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/api/services/storage.py b/api/services/storage.py index 4a7aa99..c1a4df1 100644 --- a/api/services/storage.py +++ b/api/services/storage.py @@ -26,6 +26,7 @@ def get_signed_url( expiration=expiry, method=method, content_type=content_type, + response_disposition="attachment", ), expiry=expiry, method=method, diff --git a/worker/main.py b/worker/main.py index 8ce70d2..1845f26 100644 --- a/worker/main.py +++ b/worker/main.py @@ -86,9 +86,9 @@ def update_job_status(result: JobResult): method="POST", path="/process", json={ - "id": "df9c2228-143e-452f-a7be-744a85fad351", - "logo_gcs_path": "gs://code_challenge_logo_detection/uploads/df9c2228-143e-452f-a7be-744a85fad351/image/neurons_logo.jpg", - "video_gcs_path": "gs://code_challenge_logo_detection/uploads/df9c2228-143e-452f-a7be-744a85fad351/video/Video_1.mp4", + "id": "27646cab-6cea-486e-8412-516727d44e7f", + "logo_gcs_path": "gs://code_challenge_logo_detection/uploads/27646cab-6cea-486e-8412-516727d44e7f/image/neurons_logo.jpg", + "video_gcs_path": "gs://code_challenge_logo_detection/uploads/27646cab-6cea-486e-8412-516727d44e7f/video/Video_1.mp4", }, ) env = builder.get_environ() From 657c692bf237ac6fd950c9109fbf67587e6db8b0 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 18:13:38 +0200 Subject: [PATCH 14/19] deploy worker --- .github/workflows/deploy.yml | 30 ++++++++ worker/log.py | 6 ++ worker/main.py | 31 +++------ worker/requirements.txt | 13 ++-- worker/sift.py | 130 +++++++++++++++++++++-------------- worker/storage.py | 8 ++- 6 files changed, 131 insertions(+), 87 deletions(-) create mode 100644 worker/log.py diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 8cd5ca1..d3a3b4f 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -62,3 +62,33 @@ jobs: echo "Service URL: $(gcloud run services describe ${{ env.API_SERVICE_NAME }} \ --region ${{ env.REGION }} \ --format 'value(status.url)')" + + deploy-cloud-function-worker: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Authenticate to Google Cloud + uses: google-github-actions/auth@v2 + with: + credentials_json: ${{ secrets.GCP_SA_KEY }} + + - name: Set up Cloud SDK + uses: google-github-actions/setup-gcloud@v2 + + - name: Deploy Cloud Function + working-directory: ./worker + run: | + gcloud functions deploy video-logo-worker \ + --gen2 \ + --runtime python312 \ + --entry-point process_video \ + --source . \ + --memory 512MB \ + --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ + --region europe-west1 \ + --timeout 60m \ + --trigger-http \ + --allow-unauthenticated diff --git a/worker/log.py b/worker/log.py new file mode 100644 index 0000000..f675f3e --- /dev/null +++ b/worker/log.py @@ -0,0 +1,6 @@ +import logging + +logging.basicConfig( + level=logging.INFO, +) +LOGGER = logging.getLogger(__name__) diff --git a/worker/main.py b/worker/main.py index 1845f26..21f8cbe 100644 --- a/worker/main.py +++ b/worker/main.py @@ -3,11 +3,14 @@ from google.cloud import storage from pydantic import BaseModel, ValidationError import requests + from typing import Dict from pathlib import Path + from sift import detect_logo_with_sift from models import JobStatus, JobResult from storage import download_from_gcs, upload_to_gcs +from log import LOGGER BUCKET_NAME = "code_challenge_logo_detection" storage_client = storage.Client(project="jing") @@ -26,7 +29,7 @@ def process_video(request: Request): try: request = ProcessVideoRequest(**data) except ValidationError as e: - return {"error": "Invalid request", "details": e.errors()}, 400 + return {"error": "invalid request", "details": e.errors()}, 400 result = JobResult( id=request.id, @@ -34,6 +37,7 @@ def process_video(request: Request): logo_gcs_path=request.logo_gcs_path, status=JobStatus.RUNNING, ) + local_output_path = get_local_path(request.video_gcs_path, request.id) try: @@ -46,13 +50,13 @@ def process_video(request: Request): ) except Exception as e: error = str(e) + LOGGER.error(f"error processing video: {error}") result.status = JobStatus.FAILED result.error = error update_job_status(result) return result - blob_name = upload_to_gcs(local_output_path, request.id) - result.output_gcs_path = blob_name + result.output_gcs_path = upload_to_gcs(local_output_path, request.id) result.status = JobStatus.SUCCESS update_job_status(result) @@ -74,24 +78,5 @@ def update_job_status(result: JobResult): }, ) if resp.status_code != 200: - raise Exception(f"Failed to update job status: {resp.status_code} {resp.text}") + raise Exception(f"failed to update job status: {resp.status_code} {resp.text}") return - - -if __name__ == "__main__": - - from werkzeug.test import EnvironBuilder - - builder = EnvironBuilder( - method="POST", - path="/process", - json={ - "id": "27646cab-6cea-486e-8412-516727d44e7f", - "logo_gcs_path": "gs://code_challenge_logo_detection/uploads/27646cab-6cea-486e-8412-516727d44e7f/image/neurons_logo.jpg", - "video_gcs_path": "gs://code_challenge_logo_detection/uploads/27646cab-6cea-486e-8412-516727d44e7f/video/Video_1.mp4", - }, - ) - env = builder.get_environ() - req = Request(env) -response = process_video(req) -print(response) diff --git a/worker/requirements.txt b/worker/requirements.txt index fa1aa91..b16ccb7 100644 --- a/worker/requirements.txt +++ b/worker/requirements.txt @@ -1,8 +1,5 @@ -functions-framework -flask -google-cloud-storage==3.4.0 -pydantic==2.12.0 -requests -opencv-python-headless -numpy - +functions-framework==3.4.0 +google-cloud-storage==2.9.0 +pydantic==2.11.7 +requests==2.28.1 +opencv-python==4.12.0.88 \ No newline at end of file diff --git a/worker/sift.py b/worker/sift.py index 3271671..a3557ba 100644 --- a/worker/sift.py +++ b/worker/sift.py @@ -1,5 +1,9 @@ import cv2 import numpy as np +from typing import List +from log import LOGGER + +FLANN_INDEX_KDTREE = 1 def detect_logo_with_sift( @@ -9,31 +13,16 @@ def detect_logo_with_sift( min_matches: int = 15, ratio_threshold: float = 0.75, ): - - logo = cv2.imread(logo_path) - if logo is None: - raise ValueError(f"Could not load logo: {logo_path}") - - logo_gray = cv2.cvtColor(logo, cv2.COLOR_BGR2GRAY) - detector = cv2.SIFT_create() - key_point, descriptor = detector.detectAndCompute(logo_gray, None) - - if descriptor is None or len(key_point) < 4: - raise ValueError("Not enough features in logo. Try a more detailed logo image.") + logo_gray, key_point, descriptor = load_logo(detector, logo_path) + cap = load_video(video_path) + fps, width, height, total_frames = get_video_info(cap) - FLANN_INDEX_KDTREE = 1 index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) search_params = dict(checks=50) matcher = cv2.FlannBasedMatcher(index_params, search_params) - cap = cv2.VideoCapture(video_path) - if not cap.isOpened(): - raise ValueError(f"Could not open video: {video_path}") - - fps, width, height, total_frames = get_video_info(cap) - fourcc = cv2.VideoWriter_fourcc(*"mp4v") out = cv2.VideoWriter(output_path, fourcc, fps, (width, height)) @@ -48,7 +37,6 @@ def detect_logo_with_sift( frame_count += 1 frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) - frame_kp, frame_desc = detector.detectAndCompute(frame_gray, None) if frame_desc is not None and len(frame_kp) >= 4: @@ -65,48 +53,84 @@ def detect_logo_with_sift( frames_with_logo += 1 total_matches += len(good_matches) - src_pts = np.float32( - [key_point[m.queryIdx].pt for m in good_matches] - ).reshape(-1, 1, 2) + frame = draw_logo_outline( + frame=frame, + logo_gray=logo_gray, + key_logo=key_point, + key_frame=frame_kp, + good_matches=good_matches, + ) + + out.write(frame) - dst_pts = np.float32( - [frame_kp[m.trainIdx].pt for m in good_matches] - ).reshape(-1, 1, 2) + if frame_count % 30 == 0: + progress = (frame_count / total_frames) * 100 + LOGGER.info(f"Progress: {progress:.1f}% ({frame_count}/{total_frames})") - M, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) + cap.release() + out.release() - if M is not None: - h, w = logo_gray.shape - logo_corners = np.float32([[0, 0], [w, 0], [w, h], [0, h]]).reshape( - -1, 1, 2 - ) - frame_corners = cv2.perspectiveTransform(logo_corners, M) +def load_logo( + detector: cv2.SIFT, + logo_path: str, +) -> tuple[np.ndarray, List[cv2.KeyPoint], List[cv2.KeyPoint]]: + logo = cv2.imread(logo_path) + if logo is None: + raise ValueError(f"Could not load logo: {logo_path}") - frame = cv2.polylines( - frame, [np.int32(frame_corners)], True, (0, 255, 0), 3 - ) + logo_gray = cv2.cvtColor(logo, cv2.COLOR_BGR2GRAY) + key_point, descriptor = detector.detectAndCompute(logo_gray, None) - x, y = int(frame_corners[0][0][0]), int(frame_corners[0][0][1]) - label = f"Logo: {len(good_matches)} matches" - cv2.putText( - frame, - label, - (x, max(y - 10, 20)), - cv2.FONT_HERSHEY_SIMPLEX, - 0.7, - (0, 255, 0), - 2, - ) + if descriptor is None or len(key_point) < 4: + raise ValueError("Not enough features in logo. Try a more detailed logo image.") - out.write(frame) + return logo_gray, key_point, descriptor - if frame_count % 30 == 0: - progress = (frame_count / total_frames) * 100 - print(f"Progress: {progress:.1f}% ({frame_count}/{total_frames})") - cap.release() - out.release() +def load_video(video_path: str) -> cv2.VideoCapture: + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + raise ValueError(f"Could not open video: {video_path}") + return cap + + +def draw_logo_outline( + frame: np.ndarray, + logo_gray: np.ndarray, + key_logo: np.ndarray, + key_frame: np.ndarray, + good_matches: List[cv2.DMatch], +) -> np.ndarray: + src_pts = np.float32([key_logo[m.queryIdx].pt for m in good_matches]).reshape( + -1, 1, 2 + ) + dst_pts = np.float32([key_frame[m.trainIdx].pt for m in good_matches]).reshape( + -1, 1, 2 + ) + + M, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) + if M is None: + return frame + + h, w = logo_gray.shape + logo_corners = np.float32([[0, 0], [w, 0], [w, h], [0, h]]).reshape(-1, 1, 2) + frame_corners = cv2.perspectiveTransform(logo_corners, M) + + frame = cv2.polylines(frame, [np.int32(frame_corners)], True, (0, 255, 0), 3) + + x, y = int(frame_corners[0][0][0]), int(frame_corners[0][0][1]) + label = f"Logo: {len(good_matches)} matches" + cv2.putText( + frame, + label, + (x, max(y - 10, 20)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 255, 0), + 2, + ) + return frame def get_video_info(cap: cv2.VideoCapture) -> tuple[int, int, int, int]: @@ -115,7 +139,7 @@ def get_video_info(cap: cv2.VideoCapture) -> tuple[int, int, int, int]: height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) - print(f"\nVideo: {width}x{height}, {fps} FPS, {total_frames} frames") + LOGGER.info(f"\nVideo: {width}x{height}, {fps} FPS, {total_frames} frames") return fps, width, height, total_frames diff --git a/worker/storage.py b/worker/storage.py index 39424e1..dc5ce86 100644 --- a/worker/storage.py +++ b/worker/storage.py @@ -1,6 +1,8 @@ from google.cloud import storage from pathlib import Path +from log import LOGGER + storage_client = storage.Client(project="jing-264314") @@ -19,9 +21,9 @@ def download_from_gcs(gcs_url, id: str): bucket = storage_client.bucket(bucket_name) blob = bucket.blob(blob_path) - print(f"Downloading {gcs_url} to {local_path}") + LOGGER.info(f"downloading {gcs_url}") blob.download_to_filename(local_path) - print(f"Downloaded {gcs_url} to {local_path}") + LOGGER.info(f"downloaded to {local_path}") return local_path @@ -32,5 +34,5 @@ def upload_to_gcs(local_path: str, id: str) -> str: ) blob.upload_from_filename(local_path) gcs_url = f"gs://{blob.bucket.name}/{blob.name}" - print(f"Result uploaded to {gcs_url}") + LOGGER.info(f"result uploaded to {gcs_url}") return gcs_url From 33378472feec975a164f33ac01509293213180fa Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 19:06:28 +0200 Subject: [PATCH 15/19] trigger cloud function to process video --- api/main.py | 26 ++++++++++++++++++++++---- api/models.py | 5 +++++ api/requirements.txt | 1 + worker/main.py | 8 +------- worker/models.py | 6 ++++++ 5 files changed, 35 insertions(+), 11 deletions(-) diff --git a/api/main.py b/api/main.py index ccc0999..be865e8 100644 --- a/api/main.py +++ b/api/main.py @@ -1,6 +1,7 @@ from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import FastAPI, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware +import httpx from datetime import datetime, timezone from logging import getLogger, INFO @@ -15,7 +16,7 @@ MissingLogoFileException, InvalidGCSURLException, ) -from models import JobStatus +from models import JobStatus, ProcessVideoPayload from input import UploadInput, JobPatchInput from output import JobOutput, CreateJobOutput @@ -84,12 +85,18 @@ async def get_job(id: str) -> JobOutput: @api.post("/jobs/{id}/run", response_model=JobOutput) -async def run_job(id: str) -> JobOutput: - # TODO trigger cloud function +async def run_job(id: str, background_tasks: BackgroundTasks) -> JobOutput: job = await get(id) if job is None: raise JobNotFoundException(id) + + payload = ProcessVideoPayload( + id=id, + video_gcs_path=job.video_gcs_url, + logo_gcs_path=job.logo_gcs_url, + ) + background_tasks.add_task(trigger_cloud_function, payload.model_dump()) job.status = JobStatus.RUNNING job.error = None job.output_gcs_url = None @@ -138,3 +145,14 @@ def validate_upload_input(input: UploadInput): if not logo_file_name.lower().endswith((".jpg")): raise InvalidImageFileExtensionException(logo_file_name) return + + +async def trigger_cloud_function(data: dict): + async with httpx.AsyncClient(timeout=0.1) as client: + try: + await client.post( + "https://europe-west1-jing-264314.cloudfunctions.net/video-logo-worker", + json=data, + ) + except Exception as e: + LOGGER.error(f"error triggering cloud function: {e}") diff --git a/api/models.py b/api/models.py index d8c94dc..0b9b895 100644 --- a/api/models.py +++ b/api/models.py @@ -32,3 +32,8 @@ class JobStatus(str, Enum): RUNNING = "running" SUCCESS = "success" FAILED = "failed" + +class ProcessVideoPayload(BaseModel): + id: str + video_gcs_path: str + logo_gcs_path: str diff --git a/api/requirements.txt b/api/requirements.txt index e9dafc2..fe51a88 100644 --- a/api/requirements.txt +++ b/api/requirements.txt @@ -4,3 +4,4 @@ google-cloud-storage==3.4.0 pydantic==2.12.0 urllib3==2.5.0 uvicorn==0.37.0 +httpx==0.28.1 \ No newline at end of file diff --git a/worker/main.py b/worker/main.py index 21f8cbe..7f85681 100644 --- a/worker/main.py +++ b/worker/main.py @@ -8,7 +8,7 @@ from pathlib import Path from sift import detect_logo_with_sift -from models import JobStatus, JobResult +from models import JobStatus, JobResult, ProcessVideoRequest from storage import download_from_gcs, upload_to_gcs from log import LOGGER @@ -16,12 +16,6 @@ storage_client = storage.Client(project="jing") -class ProcessVideoRequest(BaseModel): - id: str - video_gcs_path: str - logo_gcs_path: str - - @functions_framework.http def process_video(request: Request): diff --git a/worker/models.py b/worker/models.py index a2508a8..9a78c78 100644 --- a/worker/models.py +++ b/worker/models.py @@ -4,6 +4,12 @@ from enum import Enum +class ProcessVideoRequest(BaseModel): + id: str + video_gcs_path: str + logo_gcs_path: str + + class JobStatus(str, Enum): RUNNING = "running" SUCCESS = "success" From 05f90a4d28f93d328807426a0b6efbc6c0df10fc Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 19:15:09 +0200 Subject: [PATCH 16/19] increase memory --- .github/workflows/deploy.yml | 2 +- api/README.md | 4 ++++ worker/README.md | 19 +++++++++++++++++++ 3 files changed, 24 insertions(+), 1 deletion(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index d3a3b4f..d5908da 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -86,7 +86,7 @@ jobs: --runtime python312 \ --entry-point process_video \ --source . \ - --memory 512MB \ + --memory 4GB \ --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ --region europe-west1 \ --timeout 60m \ diff --git a/api/README.md b/api/README.md index c6c599f..e10e383 100644 --- a/api/README.md +++ b/api/README.md @@ -6,6 +6,10 @@ FastAPI application for video logo processing with GCP integration. 1. **Download GCP credentials** and save as `adc.json` in `/api` directory. +```bash +export GOOGLE_APPLICATION_CREDENTIALS=/path/to/adc.json +``` + 2. **Build and run**: ```bash diff --git a/worker/README.md b/worker/README.md index 496e81b..3eae757 100644 --- a/worker/README.md +++ b/worker/README.md @@ -1 +1,20 @@ # video-logo-tracker worker + +Cloud Function for video logo processing. + +## Run Worker Locally with Docker + +1. **Download GCP credentials** and save as `adc.json` in `/worker` directory. + +```bash +export GOOGLE_APPLICATION_CREDENTIALS=/path/to/adc.json +``` + +2. **Build and run**: + +```bash +cd worker +uv venv +uv pip install -r requirements.txt +uv run python main.py +``` From dfda56766ae2b74f9a0e02296d2b4b4c19c444c1 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 19:29:09 +0200 Subject: [PATCH 17/19] fix cloud funtion return value --- .github/workflows/deploy.yml | 3 ++- README.md | 19 ++++++++++++++++++- worker/main.py | 4 ++-- 3 files changed, 22 insertions(+), 4 deletions(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index d5908da..2c5d72d 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -86,7 +86,8 @@ jobs: --runtime python312 \ --entry-point process_video \ --source . \ - --memory 4GB \ + --memory 8GB \ + --cpu 4 \ --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ --region europe-west1 \ --timeout 60m \ diff --git a/README.md b/README.md index 5d0a180..fe3f7b8 100644 --- a/README.md +++ b/README.md @@ -1 +1,18 @@ -# video-logo-tracker \ No newline at end of file +# video-logo-tracker (code challenge) + +This is a code challenge for a video logo tracker. + +It consists of two parts: + +1. API +2. Worker + +## API + +The API is a FastAPI application deployed on GCP cloud run that allows you to upload a video and a logo, and then process the video to detect the logo. + +## Worker + +The Worker is a Cloud Function deployed on GCP that processes the video to detect the logo using the SIFT algorithm. + +Visit the APP \ No newline at end of file diff --git a/worker/main.py b/worker/main.py index 7f85681..76ee919 100644 --- a/worker/main.py +++ b/worker/main.py @@ -48,13 +48,13 @@ def process_video(request: Request): result.status = JobStatus.FAILED result.error = error update_job_status(result) - return result + return result.model_dump(), 200 result.output_gcs_path = upload_to_gcs(local_output_path, request.id) result.status = JobStatus.SUCCESS update_job_status(result) - return result + return result.model_dump(), 200 def get_local_path(video_gcs_path: str, id: str) -> str: From 377820ad9eab693da6d1e25e132dc7972c66df7e Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 19:33:26 +0200 Subject: [PATCH 18/19] set max instance --- .github/workflows/deploy.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 2c5d72d..351d67a 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -88,6 +88,7 @@ jobs: --source . \ --memory 8GB \ --cpu 4 \ + --max-instances 4 \ --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ --region europe-west1 \ --timeout 60m \ From a8e610bce2b043baa347b82e279040d6ca6318d9 Mon Sep 17 00:00:00 2001 From: JingLin0 Date: Sat, 11 Oct 2025 19:39:55 +0200 Subject: [PATCH 19/19] speed up --- .github/workflows/deploy.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 351d67a..a4825ea 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -86,8 +86,8 @@ jobs: --runtime python312 \ --entry-point process_video \ --source . \ - --memory 8GB \ - --cpu 4 \ + --memory 16GB \ + --cpu 8 \ --max-instances 4 \ --service-account neurons-cloud-function@jing-264314.iam.gserviceaccount.com \ --region europe-west1 \