Code Reference, Core & Search¶

Auto-generated from source for the shared core models and the legacy search stack.

Core¶

common¶

common ¶

Item `dataclass` ¶

Item(
    id: int,
    title: str,
    public_url: str,
    text: Optional[str] = "",
    creator: Optional[str] = "",
    locations: Optional[List[Dict[str, Any]]] = list(),
    geo_metadata: Optional[Dict[str, Any]] = dict(),
    time_metadata: Optional[Dict[str, Any]] = dict(),
    files_url: List[str] = list(),
)

char_count `property` ¶

char_count: int

Returns the length of text_all in characters.

word_count `property` ¶

word_count: int

Returns the word count of text_all.

clean_payload_field ¶

clean_payload_field(data: Any) -> Any

Converts data containing NumPy types to native Python types.

Source code in ai-engine\src\ai_engine\common.py

def clean_payload_field(data: Any) -> Any:
    """Converts data containing NumPy types to native Python types."""
    if isinstance(data, (np.ndarray, pd.Series)):
        return data.tolist()

hit_to_item ¶

hit_to_item(
    hit: Any,
    source: SearchType,
    query_text: Optional[str] = None,
) -> dict[str, Any]

Convert a Qdrant point to a small, standardized dict without assuming anything about the payload structure.

Source code in ai-engine\src\ai_engine\common.py

def hit_to_item(
    hit: Any,          # Qdrant ScoredPoint or Record
    source: SearchType,
    query_text: Optional[str] = None,
) -> dict[str, Any]:
    """
    Convert a Qdrant point to a small, standardized dict without
    assuming anything about the payload structure.
    """
    payload = hit.payload or {}
    score = getattr(hit, "score", None)

    highlight = None
    if query_text:
        # You can decide later how/if to use this
        text_for_highlight = payload.get("text") or ""
        highlight = highlight_search_match(text_for_highlight, query_text)

    return {
        "id": hit.id,
        "source": source,
        "score": score,
        "payload": payload,
        "highlight": highlight,
    }

limit_text ¶

limit_text(text: str, lim: int = 80)

limit_text("hello world", 5) 'hello...'

limit_text("hello world", 100) 'hello world'

:param text: Text to limit :param lim: Max length :return: Limited text

Source code in ai-engine\src\ai_engine\common.py

def limit_text(text: str, lim: int = 80):
    """
    >>> limit_text("hello world", 5)
    'hello...'

    >>> limit_text("hello world", 100)
    'hello world'

    :param text: Text to limit
    :param lim: Max length
    :return: Limited text
    """
    if len(text) > lim:
        return text[:lim] + "..."
    return text

highlight_search_match ¶

highlight_search_match(
    text: str, query: str, before="<b>", after="</b>"
)

highlight_search_match("hello world", "world") 'hello world'

highlight_search_match("hello world", "hello") 'hello world'

highlight_search_match("hello world", "hell") 'hello world'

highlight_search_match("Hello world", "hell") 'Hello world'

highlight_search_match("hello world", "foo") 'hello world'

highlight_search_match("hello world", "ello") 'hello world'

:param text: Found string :param query: Search query :param before: Tag before match :param after: Tag after match :return: Highlighted string

Source code in ai-engine\src\ai_engine\common.py

def highlight_search_match(text: str, query: str, before="<b>", after="</b>"):
    """
    >>> highlight_search_match("hello world", "world")
    'hello <b>world</b>'

    >>> highlight_search_match("hello world", "hello")
    '<b>hello</b> world'

    >>> highlight_search_match("hello world", "hell")
    '<b>hell</b>o world'

    >>> highlight_search_match("Hello world", "hell")
    '<b>Hell</b>o world'

    >>> highlight_search_match("hello world", "foo")
    'hello world'

    >>> highlight_search_match("hello world", "ello")
    'hello world'


    :param text: Found string
    :param query: Search query
    :param before: Tag before match
    :param after: Tag after match
    :return: Highlighted string
    """
    # Replace matches only on word boundaries
    return re.compile(r"\b(" + re.escape(query) + ")", re.IGNORECASE).sub(before + '\\1' + after, text)

config¶

config ¶

user_state¶

user_state ¶

UserState ¶

UserState(config: UserStateConfig = UserStateConfig())

Source code in ai-engine\src\ai_engine\user_state.py

def __init__(self, config: UserStateConfig = UserStateConfig()):
    self.config = config

is_interaction_successful ¶

is_interaction_successful(
    dwell_time: float, estimated_reading_time: float
) -> bool

Determine if user interaction is successful based on dwell time compared to estimated reading time.

Source code in ai-engine\src\ai_engine\user_state.py

def is_interaction_successful(self, dwell_time: float, estimated_reading_time: float) -> bool:
    """
    Determine if user interaction is successful based on dwell time
    compared to estimated reading time.
    """
    return dwell_time >= estimated_reading_time

compute_reading_time ¶

compute_reading_time(
    content_length_words: int, has_image: bool = False
) -> float

Estimate reading time for given content length in words. If content has an image, add extra fixed time.

Source code in ai-engine\src\ai_engine\user_state.py

def compute_reading_time(self, content_length_words: int, has_image: bool = False) -> float:
    """
    Estimate reading time for given content length in words.
    If content has an image, add extra fixed time.
    """
    base_time = content_length_words / self.config.reading_speed_wps
    if has_image:
        base_time += self.config.img_extra_fixed_time
    return base_time

projection_builder¶

projection_builder ¶

ProjectionBuilder ¶

ProjectionBuilder(collection_name: str)

Source code in ai-engine\src\ai_engine\projection_builder.py

def __init__(self, collection_name: str):
    self.user_state = UserState()
    self.db_engine = DB_Interface()
    self.common_searcher = CommonSearch(collection_name=collection_name)

get_user_projection ¶

get_user_projection(user_id: int) -> DataFrame

Compute user projection embedding based on recent events.

Source code in ai-engine\src\ai_engine\projection_builder.py

def get_user_projection(self, user_id: int) -> pd.DataFrame:
    """
    Compute user projection embedding based on recent events.
    """
    # Fetch user events
    events_df = self.db_engine.fetch_events(user_id=int(user_id))
    logger.info(events_df.dtypes)
    item_details = self._get_item_details(item_id=events_df['item_id'].astype(int).unique().tolist())

    # For each row in events_df, compute estimated reading time and success flag
    for idx, event in events_df.iterrows():
        item_id = int(event['item_id'])
        word_count = item_details[item_id]['word_count']
        has_image = item_details[item_id]['has_image']
        estimated_time = self.user_state.compute_reading_time(content_length_words=word_count, has_image=has_image)
        events_df.at[idx, 'estimated_reading_time'] = estimated_time
        is_successful = self.user_state.is_interaction_successful(
            dwell_time=event['dwell_seconds'],
            estimated_reading_time=estimated_time
        )
        events_df.at[idx, 'is_successful'] = is_successful

    return events_df

get_user_profile_as_text ¶

get_user_profile_as_text(user_id: int) -> str

Retrieve the user profile as a string

Source code in ai-engine\src\ai_engine\projection_builder.py

def get_user_profile_as_text(self, user_id: int) -> str:
    """
    Retrieve the user profile as a string
    """
    user = self.db_engine.fetch_user(user_id=int(user_id)).to_dict(orient='records')[0]
    return self._user_to_query_text(User(**user))

db_interface¶

db_interface ¶

DB_Interface ¶

DB_Interface()

Source code in ai-engine\src\ai_engine\db_interface.py

def __init__(self):
    # Create engine
    self.engine = create_engine(
        url=URL.create(
            database=DB_NAME,
            host=DB_HOST,
            username=DB_USER,
            password=DB_PASSWORD,
            port=DB_PORT,
            drivername=DB_DRIVERNAME),
        pool_pre_ping=True
    )

register_session ¶

register_session(session: Session) -> dict

Creates a new device row (if needed) and a new session row using the input schema.

Source code in ai-engine\src\ai_engine\db_interface.py

def register_session(self, session: Session) -> dict:
    """
    Creates a new device row (if needed) and a new session row using the input schema.
    """
    new_sess_token = secrets.token_urlsafe(32) 
    expires_at = datetime.now() + timedelta(days=session.expires_in_days)

    try:
        with self.engine.begin() as conn: # Transaction start

            # --- A. Get or Create Device (Logic using data.device_token) ---
            sql_query = "SELECT id FROM device WHERE device_id_token = :t"
            dev_id = conn.execute(text(sql_query), {"t": session.device_token}).scalar()

            if not dev_id:
                # Insert new device row
                sql_create_dev = """
                INSERT INTO device (user_id, device_id_token, user_agent, last_ip)
                VALUES (:uid, :token, :ua, :ip) RETURNING id;
                """
                dev_id = conn.execute(text(sql_create_dev), {
                    "uid": session.user_id, 
                    "token": session.device_token, 
                    "ua": session.user_agent, 
                    "ip": session.ip
                }).scalar()
                logger.info(f"Created NEW device with UUID: {dev_id}")
            else:
                # Update last seen info for known device
                conn.execute(text("UPDATE device SET last_seen_at = NOW(), last_ip = :ip WHERE id = :id"), 
                            {"ip": session.ip, "id": dev_id})
                logger.info(f"Found EXISTING device with UUID: {dev_id}. Updated last_seen_at.")

            # --- B. Insert Session (Logic using data.user_id, dev_id, etc.) ---
            sql_session = """
            INSERT INTO session (user_id, device_id, session_token, ip, user_agent, expires_at)
            VALUES (:uid, :did, :tok, :ip, :ua, :exp) 
            RETURNING id;
            """
            sess_uuid = conn.execute(text(sql_session), {
                "uid": session.user_id, 
                "did": dev_id, 
                "tok": new_sess_token, 
                "ip": session.ip, 
                "ua": session.user_agent, 
                "exp": expires_at
            }).scalar()

        logger.info(
            f"Successful insert of Session. User ID: {session.user_id or 'GUEST'} "
            f"| Session ID: {sess_uuid} | Expires: {expires_at.strftime('%Y-%m-%d %H:%M')}"
        )

        return {
            "status": "ok", 
            "session_id": sess_uuid,     
            "session_token": new_sess_token
        }

    except Exception as e:
        logger.exception(f"Session creation failed: {e}")
        return {"status": "error"}

narrative¶

narrative ¶

OpenRouterAdapter ¶

OpenRouterAdapter(model: str)

Acts as a drop-in replacement for ollama.Client() for testing. Uses the OpenAI SDK pointed at the OpenRouter API.