Mina Choi
14b6e95f9d
사업장 API 가 전부 "요청자 = 소유자" 로 걸러져서 관리자가 남의 빌더·미니블로그를 열면 PLACE_NOT_FOUND 였다. 운영 중 사업장 정리도 사장님 계정으로 들어가야 했다. - validator/dependencies: DEVELOPER 가 place_id 경로를 부르면 그 사업장 소유자 신원으로 처리 — [admin-access] 로그 - ops/sites: search(상호·소유자 이메일·이름·아이디) · category 필터, 건수도 같은 조건 - OpsSitesPage: 검색창 · 업종 칩 · 행마다 빌더·미니블로그·삭제(공개 중이면 발행 내린 뒤 삭제) - SitesPage: 더보기에서 미니블로그 관리 · 예약요청 관리 제거 테스트 4건 추가(test_admin_access). 전체 900 passed · 145 failed(변경 전과 같은 파일: agent_runtime·gemini 등) frontend tsc·eslint 통과 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
3.9 KiB
Python
87 lines
3.9 KiB
Python
from fastapi import Depends
|
|
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.database.model.models import places, users
|
|
from common.enums import DBWRType, ErrorType, UserRole
|
|
from common.models.gmodel import PageParams
|
|
from crud.site_crud import ISiteCRUD, SiteCRUD
|
|
from crud.user_crud import IUserCRUD, UserCRUD
|
|
from router.v1.ops.protocol import OpsSiteData, OpsUserData, Res_OpsSites, Res_OpsUsers
|
|
|
|
|
|
class OpsService:
|
|
"""내부 운영(개발자) 전용 조회 — 회사 스코프 없는 전 계정 사이트·유저 목록."""
|
|
|
|
def __init__(self, site_crud: ISiteCRUD = Depends(SiteCRUD), user_crud: IUserCRUD = Depends(UserCRUD)):
|
|
self.site_crud = site_crud
|
|
self.user_crud = user_crud
|
|
|
|
async def list_sites(self, pg: PageParams, search: str | None = None, category: int | None = None) -> Res_OpsSites:
|
|
res = Res_OpsSites(page=pg.page, size=pg.size)
|
|
err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
|
|
places.DBType(), DBWRType.DB_READ.value,
|
|
lambda s: self.site_crud.list_all_sites(s, pg.skip, pg.size, search, category),
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
res.sites = [
|
|
self._site_row(place, site, built_at, photo, login_id, email, name)
|
|
for place, site, built_at, photo, login_id, email, name in rows
|
|
]
|
|
res.total = total
|
|
return res
|
|
|
|
@staticmethod
|
|
def _site_row(place, site, built_at, primary_photo_url, owner_login_id, owner_email, owner_name) -> OpsSiteData:
|
|
# 재빌드 판별은 site_service._my_site_row 와 같은 규칙이어야 한다 — 다르면 화면마다 다른 답을 한다.
|
|
changed = place.content_updated_at
|
|
ever_published = site is not None and getattr(site, "published_at", None) is not None
|
|
thumbnail_url = getattr(site, "thumbnail_url", None) or (primary_photo_url if ever_published else None)
|
|
return OpsSiteData(
|
|
place_id=place.place_id,
|
|
name=place.name,
|
|
category=place.category,
|
|
place_status=place.status,
|
|
created_at=place.created_at,
|
|
site_id=getattr(site, "site_id", None),
|
|
status=getattr(site, "status", None),
|
|
domain=getattr(site, "domain", None),
|
|
published_at=getattr(site, "published_at", None),
|
|
thumbnail_url=thumbnail_url,
|
|
needs_rebuild=bool(site is not None and changed and (built_at is None or changed > built_at)),
|
|
owner_login_id=owner_login_id,
|
|
owner_email=owner_email,
|
|
owner_name=owner_name,
|
|
)
|
|
|
|
async def list_users(self, pg: PageParams, search: str | None = None) -> Res_OpsUsers:
|
|
"""USER/OWNER 만 — 개발자 계정은 목록에서 뺀다(UserRole 주석: 고객사에 존재를 노출하지 않는다)."""
|
|
res = Res_OpsUsers(page=pg.page, size=pg.size)
|
|
roles = [UserRole.USER.value, UserRole.OWNER.value]
|
|
err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
|
|
users.DBType(), DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.list_users(s, roles, pg.skip, pg.size, search),
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
res.users = [
|
|
OpsUserData(
|
|
user_id=user.user_id,
|
|
login_id=user.id,
|
|
name=user.name,
|
|
email=user.email,
|
|
contact_number=user.contact_number,
|
|
role=user.role,
|
|
status=user.status,
|
|
provider=user.provider,
|
|
created_at=user.created_at,
|
|
last_accessed_at=user.last_accessed_at,
|
|
place_count=place_count,
|
|
)
|
|
for user, place_count in rows
|
|
]
|
|
res.total = total
|
|
return res
|