Add admin debug endpoints for listings
Some checks failed
CI / Frontend Lint & Type Check (push) Has been cancelled
CI / Frontend Build (push) Has been cancelled
CI / Backend Lint (push) Has been cancelled
CI / Backend Tests (push) Has been cancelled
CI / Docker Build (push) Has been cancelled
CI / Security Scan (push) Has been cancelled
Deploy / Build & Push Images (push) Has been cancelled
Deploy / Deploy to Server (push) Has been cancelled
Deploy / Notify (push) Has been cancelled

This commit is contained in:
2025-12-16 15:11:08 +01:00
parent 82619f5506
commit dfdee7512a

View File

@ -25,6 +25,7 @@ from app.models.newsletter import NewsletterSubscriber
from app.models.tld_price import TLDPrice, TLDInfo
from app.models.auction import DomainAuction
from app.models.price_alert import PriceAlert
from app.models.listing import DomainListing
from app.services.db_backup import create_backup, list_backups
from app.services.ops_alerts import run_ops_alert_checks
from app.models.ops_alert import OpsAlertEvent
@ -1445,3 +1446,76 @@ async def set_user_subscription(
"old_tier": old_tier.value if old_tier else None,
"new_tier": tier.value,
}
# ============== Listing Debug Endpoints ==============
@router.get("/listings/debug")
async def debug_listings(
domain: Optional[str] = None,
slug: Optional[str] = None,
db: Database = None,
_: User = Depends(require_admin),
):
"""Debug listings - search by domain or slug (ignores status)."""
query = select(DomainListing)
if domain:
query = query.where(DomainListing.domain.ilike(f"%{domain}%"))
if slug:
query = query.where(DomainListing.slug.ilike(f"%{slug}%"))
query = query.order_by(desc(DomainListing.created_at)).limit(20)
result = await db.execute(query)
listings = list(result.scalars().all())
return {
"count": len(listings),
"listings": [
{
"id": l.id,
"domain": l.domain,
"slug": l.slug,
"status": l.status,
"is_verified": l.is_verified,
"verification_status": l.verification_status,
"public_url": l.public_url,
"created_at": str(l.created_at) if l.created_at else None,
"published_at": str(l.published_at) if l.published_at else None,
"user_id": l.user_id,
}
for l in listings
]
}
@router.post("/listings/{listing_id}/force-activate")
async def force_activate_listing(
listing_id: int,
db: Database = None,
_: User = Depends(require_admin),
):
"""Force-activate a listing (bypass DNS verification)."""
result = await db.execute(
select(DomainListing).where(DomainListing.id == listing_id)
)
listing = result.scalar_one_or_none()
if not listing:
raise HTTPException(status_code=404, detail="Listing not found")
listing.status = "active"
listing.is_verified = True
listing.verification_status = "verified"
listing.published_at = datetime.utcnow()
await db.commit()
return {
"status": "activated",
"listing_id": listing.id,
"domain": listing.domain,
"slug": listing.slug,
"public_url": listing.public_url,
}