boki/stock-bot
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

work on qm

Browse source
This commit is contained in:
Boki 2025-06-21 23:45:57 -04:00
parent ca1f658be6
commit 3fb9df425c
5 changed files with 93 additions and 113 deletions
Download patch file Download diff file Expand all files Collapse all files

27
apps/data-ingestion/src/handlers/qm/actions/session.action.ts
View file

@ -2,14 +2,14 @@
* QM Session Actions - Session management and creation * QM Session Actions - Session management and creation
*/ */
import type { IServiceContainer } from '@stock-bot/handlers'; import { BaseHandler } from '@stock-bot/core/handlers';
import { QM_SESSION_IDS, SESSION_CONFIG } from '../shared/config'; import { QM_SESSION_IDS, SESSION_CONFIG } from '../shared/config';
import { QMSessionManager } from '../shared/session-manager'; import { QMSessionManager } from '../shared/session-manager';
/** /**
* Check existing sessions and queue creation jobs for needed sessions * Check existing sessions and queue creation jobs for needed sessions
*/ */
export async function checkSessions(services: IServiceContainer): Promise<{ export async function checkSessions(handler: BaseHandler): Promise<{
cleaned: number; cleaned: number;
queued: number; queued: number;
message: string; message: string;
@ -19,12 +19,13 @@ export async function checkSessions(services: IServiceContainer): Promise<{
// Check which session IDs need more sessions and queue creation jobs // Check which session IDs need more sessions and queue creation jobs
let queuedCount = 0; let queuedCount = 0;
for (const sessionId of Object.values(QM_SESSION_IDS)) { for (const sessionId of Object.values(QM_SESSION_IDS)) {
console.log(`Checking session ID: ${sessionId}`);
if (sessionManager.needsMoreSessions(sessionId)) { if (sessionManager.needsMoreSessions(sessionId)) {
const currentCount = sessionManager.getSessions(sessionId).length; const currentCount = sessionManager.getSessions(sessionId).length;
const neededSessions = SESSION_CONFIG.MAX_SESSIONS - currentCount; const neededSessions = SESSION_CONFIG.MAX_SESSIONS - currentCount;
for (let i = 0; i < neededSessions; i++) { for (let i = 0; i < neededSessions; i++) {
await services.queue.getQueue('qm').add('create-session', { sessionId }); await handler.scheduleOperation('create-session', { sessionId });
services.logger.log(`Queued job to create session for ${sessionId}`); handler.services.logger.log(`Queued job to create session for ${sessionId}`);
queuedCount++; queuedCount++;
} }
} }
@ -41,29 +42,17 @@ export async function checkSessions(services: IServiceContainer): Promise<{
* Create a single session for a specific session ID * Create a single session for a specific session ID
*/ */
export async function createSingleSession( export async function createSingleSession(
services: IServiceContainer, handler: BaseHandler,
input: any input: any
): Promise<{ sessionId: string; status: string; sessionType: string }> { ): Promise<{ sessionId: string; status: string; sessionType: string }> {
const { sessionId: sessionType = 'default' } = input || {}; const { sessionId: sessionType = 'default' } = input || {};
const sessionManager = QMSessionManager.getInstance(); const sessionManager = QMSessionManager.getInstance();
// Check if we're at capacity for this session type
if (sessionManager.isAtCapacity(sessionType)) {
return {
sessionId: '',
status: 'skipped',
sessionType,
};
}
// TODO: Get actual proxy and headers from proxy service // TODO: Get actual proxy and headers from proxy service
const session = { const session = {
proxy: 'http://proxy:8080', // Placeholder // proxy: handler.services.getRandomProxy(),
headers: { headers: sessionManager.getQmHeaders(),
'User-Agent': 'Mozilla/5.0 (compatible; QMBot/1.0)',
'Accept': 'application/json'
},
successfulCalls: 0, successfulCalls: 0,
failedCalls: 0, failedCalls: 0,
lastUsed: new Date() lastUsed: new Date()

135
apps/data-ingestion/src/handlers/qm/qm.handler.ts
View file

@ -6,7 +6,6 @@ import {
type ExecutionContext, type ExecutionContext,
type IServiceContainer type IServiceContainer
} from '@stock-bot/handlers'; } from '@stock-bot/handlers';
import type { SymbolSpiderJob } from './shared/types';
@Handler('qm') @Handler('qm')
export class QMHandler extends BaseHandler { export class QMHandler extends BaseHandler {
@ -23,90 +22,90 @@ export class QMHandler extends BaseHandler {
async checkSessions(input: unknown, context: ExecutionContext): Promise<unknown> { async checkSessions(input: unknown, context: ExecutionContext): Promise<unknown> {
// Call the session maintenance action // Call the session maintenance action
const { checkSessions } = await import('./actions/session.action'); const { checkSessions } = await import('./actions/session.action');
return await checkSessions(this.services); return await checkSessions(this);
} }
@Operation('create-session') @Operation('create-session')
async createSession(input: unknown, context: ExecutionContext): Promise<unknown> { async createSession(input: unknown, context: ExecutionContext): Promise<unknown> {
// Call the individual session creation action // Call the individual session creation action
const { createSingleSession } = await import('./actions/session.action'); const { createSingleSession } = await import('./actions/session.action');
return await createSingleSession(this.services, input); return await createSingleSession(this, input);
} }
@Operation('search-symbols') // @Operation('search-symbols')
async searchSymbols(_input: unknown, _context: ExecutionContext): Promise<unknown> { // async searchSymbols(_input: unknown, _context: ExecutionContext): Promise<unknown> {
this.logger.info('Searching QM symbols with new DI pattern...'); // this.logger.info('Searching QM symbols with new DI pattern...');
try { // try {
// Check existing symbols in MongoDB // // Check existing symbols in MongoDB
const symbolsCollection = this.mongodb.collection('qm_symbols'); // const symbolsCollection = this.mongodb.collection('qm_symbols');
const symbols = await symbolsCollection.find({}).limit(100).toArray(); // const symbols = await symbolsCollection.find({}).limit(100).toArray();
this.logger.info('QM symbol search completed', { count: symbols.length }); // this.logger.info('QM symbol search completed', { count: symbols.length });
if (symbols && symbols.length > 0) { // if (symbols && symbols.length > 0) {
// Cache result for performance // // Cache result for performance
await this.cache.set('qm-symbols-sample', symbols.slice(0, 10), 1800); // await this.cache.set('qm-symbols-sample', symbols.slice(0, 10), 1800);
return { // return {
success: true, // success: true,
message: 'QM symbol search completed successfully', // message: 'QM symbol search completed successfully',
count: symbols.length, // count: symbols.length,
symbols: symbols.slice(0, 10), // Return first 10 symbols as sample // symbols: symbols.slice(0, 10), // Return first 10 symbols as sample
}; // };
} else { // } else {
// No symbols found - this is expected initially // // No symbols found - this is expected initially
this.logger.info('No QM symbols found in database yet'); // this.logger.info('No QM symbols found in database yet');
return { // return {
success: true, // success: true,
message: 'No symbols found yet - database is empty', // message: 'No symbols found yet - database is empty',
count: 0, // count: 0,
}; // };
} // }
} catch (error) { // } catch (error) {
this.logger.error('Failed to search QM symbols', { error }); // this.logger.error('Failed to search QM symbols', { error });
throw error; // throw error;
} // }
} // }
@Operation('spider-symbol-search') // @Operation('spider-symbol-search')
@QueueSchedule('0 0 * * 0', { // @QueueSchedule('0 0 * * 0', {
priority: 10, // priority: 10,
immediately: false, // immediately: false,
description: 'Comprehensive symbol search using QM API' // description: 'Comprehensive symbol search using QM API'
}) // })
async spiderSymbolSearch(payload: SymbolSpiderJob | undefined, context: ExecutionContext): Promise<unknown> { // async spiderSymbolSearch(payload: SymbolSpiderJob | undefined, context: ExecutionContext): Promise<unknown> {
// Set default payload for scheduled runs // // Set default payload for scheduled runs
const jobPayload: SymbolSpiderJob = payload || { // const jobPayload: SymbolSpiderJob = payload || {
prefix: null, // prefix: null,
depth: 1, // depth: 1,
source: 'qm', // source: 'qm',
maxDepth: 4 // maxDepth: 4
}; // };
this.logger.info('Starting QM spider symbol search', { payload: jobPayload }); // this.logger.info('Starting QM spider symbol search', { payload: jobPayload });
// Store spider job info in cache (temporary data) // // Store spider job info in cache (temporary data)
const spiderJobId = `spider:qm:${Date.now()}:${Math.random().toString(36).substr(2, 9)}`; // const spiderJobId = `spider:qm:${Date.now()}:${Math.random().toString(36).substr(2, 9)}`;
const spiderResult = { // const spiderResult = {
payload: jobPayload, // payload: jobPayload,
startTime: new Date().toISOString(), // startTime: new Date().toISOString(),
status: 'started', // status: 'started',
jobId: spiderJobId // jobId: spiderJobId
}; // };
// Store in cache with 1 hour TTL (temporary data) // // Store in cache with 1 hour TTL (temporary data)
await this.cache.set(spiderJobId, spiderResult, 3600); // await this.cache.set(spiderJobId, spiderResult, 3600);
this.logger.debug('Spider job stored in cache', { spiderJobId, ttl: 3600 }); // this.logger.debug('Spider job stored in cache', { spiderJobId, ttl: 3600 });
// Schedule follow-up processing if needed // // Schedule follow-up processing if needed
await this.scheduleOperation('search-symbols', { source: 'spider', spiderJobId }, 5000); // await this.scheduleOperation('search-symbols', { source: 'spider', spiderJobId }, 5000);
return { // return {
success: true, // success: true,
message: 'QM spider search initiated', // message: 'QM spider search initiated',
spiderJobId // spiderJobId
}; // };
} // }
} }

17
apps/data-ingestion/src/handlers/qm/shared/config.ts
View file

@ -2,7 +2,6 @@
* Shared configuration for QM operations * Shared configuration for QM operations
*/ */
import { getRandomUserAgent } from '@stock-bot/http';
// QM Session IDs for different endpoints // QM Session IDs for different endpoints
export const QM_SESSION_IDS = { export const QM_SESSION_IDS = {
@ -28,8 +27,6 @@ export const QM_CONFIG = {
BASE_URL: 'https://app.quotemedia.com', BASE_URL: 'https://app.quotemedia.com',
AUTH_PATH: '/auth/g/authenticate/dataTool/v0/500', AUTH_PATH: '/auth/g/authenticate/dataTool/v0/500',
LOOKUP_URL: 'https://app.quotemedia.com/datatool/lookup.json', LOOKUP_URL: 'https://app.quotemedia.com/datatool/lookup.json',
ORIGIN: 'https://www.quotemedia.com',
REFERER: 'https://www.quotemedia.com/',
} as const; } as const;
// Session management settings // Session management settings
@ -40,17 +37,3 @@ export const SESSION_CONFIG = {
SESSION_TIMEOUT: 10000, // 10 seconds SESSION_TIMEOUT: 10000, // 10 seconds
API_TIMEOUT: 15000, // 15 seconds API_TIMEOUT: 15000, // 15 seconds
} as const; } as const;
/**
* Generate standard QM headers
*/
export function getQmHeaders(): Record<string, string> {
return {
'User-Agent': getRandomUserAgent(),
Accept: '*/*',
'Accept-Language': 'en',
'Sec-Fetch-Mode': 'cors',
Origin: QM_CONFIG.ORIGIN,
Referer: QM_CONFIG.REFERER,
};
}

14
apps/data-ingestion/src/handlers/qm/shared/session-manager.ts
View file

@ -2,8 +2,9 @@
* QM Session Manager - Centralized session state management * QM Session Manager - Centralized session state management
*/ */
import type { QMSession } from './types'; import { getRandomUserAgent } from '@stock-bot/services/http';
import { QM_SESSION_IDS, SESSION_CONFIG } from './config'; import { QM_SESSION_IDS, SESSION_CONFIG } from './config';
import type { QMSession } from './types';
export class QMSessionManager { export class QMSessionManager {
private static instance: QMSessionManager | null = null; private static instance: QMSessionManager | null = null;
@ -83,6 +84,17 @@ export class QMSessionManager {
return removedCount; return removedCount;
} }
getQmHeaders(): Record<string, string> {
return {
'User-Agent': getRandomUserAgent(),
Accept: '*/*',
'Accept-Language': 'en',
'Sec-Fetch-Mode': 'cors',
Origin: 'https://www.quotemedia.com',
Referer: 'https://www.quotemedia.com/',
};
}
/** /**
* Check if more sessions are needed for a session ID * Check if more sessions are needed for a session ID
*/ */

11
libs/core/handlers/src/base/BaseHandler.ts
View file

@ -1,7 +1,7 @@
import { getLogger } from '@stock-bot/logger'; import { getLogger } from '@stock-bot/logger';
import type { IHandler, ExecutionContext } from '../types/types'; import { createJobHandler, handlerRegistry, type HandlerConfigWithSchedule } from '@stock-bot/types';
import type { IServiceContainer } from '../types/service-container'; import type { IServiceContainer } from '../types/service-container';
import { handlerRegistry, createJobHandler, type HandlerConfigWithSchedule } from '@stock-bot/types'; import type { ExecutionContext, IHandler } from '../types/types';
/** /**
* Abstract base class for all handlers with improved DI * Abstract base class for all handlers with improved DI
@ -11,7 +11,7 @@ export abstract class BaseHandler implements IHandler {
protected readonly logger; protected readonly logger;
private handlerName: string; private handlerName: string;
constructor(protected readonly services: IServiceContainer, handlerName?: string) { constructor(readonly services: IServiceContainer, handlerName?: string) {
this.logger = getLogger(this.constructor.name); this.logger = getLogger(this.constructor.name);
// Read handler name from decorator first, then fallback to parameter or class name // Read handler name from decorator first, then fallback to parameter or class name
const constructor = this.constructor as any; const constructor = this.constructor as any;
@ -65,10 +65,7 @@ export abstract class BaseHandler implements IHandler {
return await method.call(this, input, context); return await method.call(this, input, context);
} }
/** async scheduleOperation(operation: string, payload: unknown, delay?: number): Promise<void> {
* Queue helper methods - now type-safe and direct
*/
protected async scheduleOperation(operation: string, payload: unknown, delay?: number): Promise<void> {
const queue = this.services.queue.getQueue(this.handlerName); const queue = this.services.queue.getQueue(this.handlerName);
const jobData = { const jobData = {
handler: this.handlerName, handler: this.handlerName,