284 lines
10 KiB
Python
284 lines
10 KiB
Python
"""Buffer GraphQL helpers and LinkedIn job-post copy rendering.
|
||
|
||
Pure module: no FastAPI imports and no HTTPException.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
from datetime import datetime
|
||
|
||
import httpx
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
BUFFER_API = os.getenv("BUFFER_API")
|
||
BUFFER_API_URL = os.getenv("BUFFER_API_URL", "https://api.buffer.com")
|
||
BUFFER_CHANNEL_ID = os.getenv("BUFFER_CHANNEL_ID")
|
||
LINKEDIN_POST_MAX_CHARS = 3000
|
||
|
||
# Buffer's PostStatus -> the job_posts.status lifecycle. Only `sent` means the post is
|
||
# actually live on the network: the default `addToQueue` mode comes back as `scheduled`,
|
||
# so treating any successful mutation as "published" would record a post that nobody
|
||
# outside Buffer can see yet.
|
||
BUFFER_STATUS_TO_LOCAL = {
|
||
"sent": "published",
|
||
"sending": "publishing",
|
||
"scheduled": "scheduled",
|
||
"draft": "draft",
|
||
"needs_approval": "needs_approval",
|
||
"error": "failed",
|
||
}
|
||
|
||
|
||
def local_status(buffer_status) -> str:
|
||
"""Map a Buffer PostStatus onto our own. Unknown values stay uncommitted."""
|
||
return BUFFER_STATUS_TO_LOCAL.get(buffer_status or "", "scheduled")
|
||
|
||
|
||
def normalize_platform(value, aliases=None) -> str:
|
||
"""Case/punctuation-insensitive key for comparing a requested platform.
|
||
|
||
`aliases` is an optional alias→Buffer-service map (from SocialPlatform). Missing
|
||
keys fall through to the normalized key so Buffer's live channel list remains the
|
||
real allowlist — DB rows are spelling tolerance only.
|
||
"""
|
||
key = re.sub(r"[^a-z0-9]", "", str(value or "").lower())
|
||
if aliases:
|
||
return aliases.get(key, key)
|
||
return key
|
||
|
||
|
||
async def resolve_channel(platform, channels=None, aliases=None) -> dict:
|
||
"""Return the connected channel for `platform`.
|
||
|
||
Nothing is special-cased per network: the request is matched against the services
|
||
Buffer reports, so any platform Buffer supports and the account has connected will
|
||
resolve. Falls back to matching the channel's handle or display name so
|
||
"ahmedmujtababaig" works as well as "linkedin".
|
||
"""
|
||
wanted = normalize_platform(platform, aliases)
|
||
if not wanted:
|
||
raise BufferError("No platform given")
|
||
if channels is None:
|
||
channels = await list_buffer_channels()
|
||
for field in ("service", "name", "displayName"):
|
||
for channel in channels:
|
||
if normalize_platform(channel.get(field), aliases) == wanted:
|
||
return channel
|
||
available = sorted({c.get("service") for c in channels if c.get("service")})
|
||
raise BufferError(
|
||
f"No Buffer channel connected for platform {platform!r}. "
|
||
f"Connected: {', '.join(available) if available else 'none'}"
|
||
)
|
||
|
||
|
||
def parse_buffer_datetime(value):
|
||
"""Buffer sends ISO 8601 with a trailing `Z`, which fromisoformat wants as +00:00."""
|
||
if not value:
|
||
return None
|
||
try:
|
||
return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
class BufferError(RuntimeError):
|
||
def __init__(self, message: str, *, code: str | None = None):
|
||
super().__init__(message)
|
||
self.code = code
|
||
|
||
|
||
def render_job_post(payload) -> str:
|
||
title = (payload.get("title") or "").strip() or "Open Role"
|
||
location = (payload.get("location") or "").strip()
|
||
employment_type = (payload.get("employment_type") or "").strip()
|
||
experience_min = payload.get("experience_min")
|
||
experience_max = payload.get("experience_max")
|
||
requirements = [str(r).strip() for r in (payload.get("requirements") or []) if str(r).strip()]
|
||
optional_skills = [str(s).strip() for s in (payload.get("optional_skills") or []) if str(s).strip()]
|
||
description = (payload.get("description") or "").strip()
|
||
|
||
lines = [f"We're hiring: {title}", ""]
|
||
|
||
meta = []
|
||
if location:
|
||
meta.append(location)
|
||
if employment_type:
|
||
meta.append(employment_type)
|
||
if meta:
|
||
lines.append(" · ".join(meta))
|
||
lines.append("")
|
||
|
||
if experience_min is not None and experience_max is not None:
|
||
lines.append(f"Experience: {experience_min}–{experience_max} years")
|
||
lines.append("")
|
||
elif experience_min is not None:
|
||
lines.append(f"Experience: {experience_min}+ years")
|
||
lines.append("")
|
||
elif experience_max is not None:
|
||
lines.append(f"Experience: up to {experience_max} years")
|
||
lines.append("")
|
||
|
||
if requirements:
|
||
lines.append("Requirements:")
|
||
for item in requirements:
|
||
lines.append(f"• {item}")
|
||
lines.append("")
|
||
|
||
if optional_skills:
|
||
lines.append("Nice to have:")
|
||
for item in optional_skills:
|
||
lines.append(f"• {item}")
|
||
lines.append("")
|
||
|
||
if description:
|
||
lines.append(description)
|
||
lines.append("")
|
||
|
||
lines.append("Interested? Apply via our careers page or reply to this post.")
|
||
lines.append("")
|
||
|
||
tags = []
|
||
for item in requirements:
|
||
tag = re.sub(r"[^A-Za-z0-9]+", "", item)
|
||
if tag:
|
||
tags.append(f"#{tag}")
|
||
if tags:
|
||
lines.append(" ".join(tags))
|
||
|
||
text = "\n".join(lines).strip()
|
||
if len(text) > LINKEDIN_POST_MAX_CHARS:
|
||
text = text[: LINKEDIN_POST_MAX_CHARS - 1].rstrip() + "…"
|
||
return text
|
||
|
||
|
||
def build_create_post_query(text, channel_id, *, mode="addToQueue", due_at=None) -> str:
|
||
fields = [
|
||
f"text: {json.dumps(text)}",
|
||
f"channelId: {json.dumps(channel_id)}",
|
||
"schedulingType: automatic",
|
||
f"mode: {mode}",
|
||
]
|
||
if mode == "customScheduled" and due_at:
|
||
fields.append(f"dueAt: {json.dumps(due_at)}")
|
||
input_block = ",\n ".join(fields)
|
||
return (
|
||
"mutation CreatePost {\n"
|
||
" createPost(input: {\n"
|
||
f" {input_block}\n"
|
||
" }) {\n"
|
||
" ... on PostActionSuccess {\n"
|
||
" post { id text status sentAt externalLink channelService }\n"
|
||
" }\n"
|
||
" ... on MutationError { message }\n"
|
||
" }\n"
|
||
"}"
|
||
)
|
||
|
||
|
||
async def create_buffer_post(text, channel_id, *, mode="addToQueue", due_at=None) -> dict:
|
||
if not BUFFER_API:
|
||
raise RuntimeError("BUFFER_API is not configured")
|
||
query = build_create_post_query(text, channel_id, mode=mode, due_at=due_at)
|
||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||
response = await client.post(
|
||
BUFFER_API_URL,
|
||
json={"query": query},
|
||
headers={
|
||
"Authorization": f"Bearer {BUFFER_API}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
)
|
||
if response.status_code != 200:
|
||
raise httpx.HTTPStatusError(
|
||
response.text,
|
||
request=response.request,
|
||
response=response,
|
||
)
|
||
body = response.json()
|
||
errors = body.get("errors")
|
||
if errors:
|
||
first = errors[0] if isinstance(errors, list) and errors else {}
|
||
msg = first.get("message") or "Buffer GraphQL error"
|
||
code = (first.get("extensions") or {}).get("code")
|
||
raise BufferError(msg, code=code)
|
||
create_post = (body.get("data") or {}).get("createPost") or {}
|
||
if "message" in create_post and "post" not in create_post:
|
||
raise BufferError(create_post.get("message") or "Buffer mutation error")
|
||
post = create_post.get("post")
|
||
if not post or not post.get("id"):
|
||
raise BufferError("Buffer did not return a post id")
|
||
return post
|
||
|
||
|
||
async def list_buffer_channels() -> list[dict]:
|
||
if not BUFFER_API:
|
||
raise RuntimeError("BUFFER_API is not configured")
|
||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||
orgs_response = await client.post(
|
||
BUFFER_API_URL,
|
||
json={"query": "query { account { organizations { id name } } }"},
|
||
headers={
|
||
"Authorization": f"Bearer {BUFFER_API}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
)
|
||
if orgs_response.status_code != 200:
|
||
raise httpx.HTTPStatusError(
|
||
orgs_response.text,
|
||
request=orgs_response.request,
|
||
response=orgs_response,
|
||
)
|
||
orgs_body = orgs_response.json()
|
||
if orgs_body.get("errors"):
|
||
first = orgs_body["errors"][0]
|
||
raise BufferError(
|
||
first.get("message") or "Buffer GraphQL error",
|
||
code=(first.get("extensions") or {}).get("code"),
|
||
)
|
||
organizations = ((orgs_body.get("data") or {}).get("account") or {}).get("organizations") or []
|
||
channels: list[dict] = []
|
||
for org in organizations:
|
||
org_id = org.get("id")
|
||
if not org_id:
|
||
continue
|
||
channels_query = (
|
||
"query GetChannels {\n"
|
||
f' channels(input:{{organizationId:{json.dumps(org_id)}}}) {{\n'
|
||
" id name displayName service isQueuePaused\n"
|
||
" }\n"
|
||
"}"
|
||
)
|
||
channels_response = await client.post(
|
||
BUFFER_API_URL,
|
||
json={"query": channels_query},
|
||
headers={
|
||
"Authorization": f"Bearer {BUFFER_API}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
)
|
||
if channels_response.status_code != 200:
|
||
raise httpx.HTTPStatusError(
|
||
channels_response.text,
|
||
request=channels_response.request,
|
||
response=channels_response,
|
||
)
|
||
channels_body = channels_response.json()
|
||
if channels_body.get("errors"):
|
||
first = channels_body["errors"][0]
|
||
raise BufferError(
|
||
first.get("message") or "Buffer GraphQL error",
|
||
code=(first.get("extensions") or {}).get("code"),
|
||
)
|
||
for channel in (channels_body.get("data") or {}).get("channels") or []:
|
||
channels.append({
|
||
**channel,
|
||
"organization_id": org_id,
|
||
"organization_name": org.get("name"),
|
||
})
|
||
return channels
|