peptide-repo/prautoblogger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: chain per-article cron jobs to eliminate execution time ceiling

Browse Source 
The pipeline previously ran all articles sequentially in a single PHP
process. On Hostinger, this timed out at ~120s — enough for 1 article
but not 2+. Now the pipeline generates article 1 in-process, then
queues articles 2..N as individual WP-Cron events. Each article gets
its own PHP process with its own execution time budget, so we can
generate any number of articles regardless of server time limits.

Architecture:
- Pipeline_Runner: orchestrate (collect+analyze+score) + first article,
  then serialize remaining ideas to wp_options and chain cron events
- Article_Worker (new): focused single-article worker extracted from
  the pipeline loop — content gen + editorial + publish
- Executor: new on_process_article_queue() handler for chained events;
  status poller re-schedules orphaned queue jobs automatically
- Article_Idea: added to_array() for queue serialization

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
peptiderepo
2026-04-19 03:26:37 +07:00
parent c866637a95
commit ef40ce91d4
6 changed files with 465 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

332
includes/core/class-pipeline-runner.php
View File

@@ -2,41 +2,37 @@
declare(strict_types=1);
/**
* Executes the full content generation pipeline.
* Executes the content generation pipeline using a chained-job architecture.
*
* Orchestrates: Budget check → Collect source data → Analyze patterns →
* Score ideas → Generate content → Editorial review → Publish.
* Score ideas → Generate article 1 in-process → Queue articles 2..N as
* individual WP-Cron events so each gets its own PHP execution time budget.
*
* Triggered by: PRAutoBlogger::on_daily_generation() and PRAutoBlogger::on_ajax_generate_now().
* Dependencies: All core content classes (Collector, Analyzer, Scorer, Generator, Editor, Publisher).
* Why chained jobs: Hostinger kills PHP processes at ~120s. A single article
* (LLM generation + editorial + 2 images) takes ~60-90s, so one-per-process
* is safe. Running N articles sequentially in one process times out at N≥2.
*
* @see class-prautoblogger.php — Instantiates this class and calls run().
* @see core/class-cost-tracker.php — Tracks cumulative API costs throughout the run.
* @see ARCHITECTURE.md — Data flow diagram showing pipeline steps.
* Triggered by: PRAutoBlogger_Executor (daily cron + manual "Generate Now").
* Dependencies: Core content classes, Article_Worker, WP-Cron for chaining.
*
* @see class-executor.php — Cron/AJAX handlers that call us.
* @see core/class-article-worker.php — Single-article generation worker.
* @see core/class-cost-tracker.php — Tracks costs throughout the run.
* @see ARCHITECTURE.md — Data flow diagram.
*/
class PRAutoBlogger_Pipeline_Runner {
/**
* Execute the full content generation pipeline.
*
* Steps: Budget check → Collect source data → Analyze patterns →
* Score ideas → Generate content → Editorial review → Publish.
*
* Side effects: API calls (OpenRouter, Reddit), database writes, post creation.
*
* @return array{generated: int, published: int, rejected: int, cost: float} Pipeline results.
*
* @throws \RuntimeException If budget is exceeded or critical step fails.
*/
/** @var bool Whether to skip duplicate-topic detection (used for manual runs). */
/** Option key for the article generation queue. */
private const QUEUE_KEY = 'prautoblogger_article_queue';
/** WP-Cron action for chained article generation. */
public const CRON_ACTION = 'prautoblogger_generate_queued_article';
/** @var bool Whether to skip duplicate-topic detection. */
private bool $skip_dedup = false;
/**
* Allow manual runs to skip the has_similar_post() check so "Generate Now"
* always produces content even when a related article was recently published.
*
* @param bool $skip True to skip deduplication.
*
* @return $this
*/
public function set_skip_dedup( bool $skip ): self {
@@ -44,128 +40,224 @@ class PRAutoBlogger_Pipeline_Runner {
return $this;
}
/**
* Run orchestration and generate the first article in-process.
*
* Steps: Budget check → Collect → Analyze → Score → Generate article 1.
* Remaining articles are queued as chained WP-Cron events.
*
* Side effects: API calls, DB writes, post creation, cron scheduling.
*
* @return array{generated: int, published: int, rejected: int, cost: float}
* @throws \RuntimeException If budget exceeded.
*/
public function run(): array {
$cost_tracker = new PRAutoBlogger_Cost_Tracker();
// Tag all log entries in this run with a unique ID so link_generation_logs()
// can accurately attribute costs to the correct post in batch runs.
$cost_tracker->set_run_id( wp_generate_uuid4() );
// Hard stop if monthly budget is exceeded.
if ( $cost_tracker->is_budget_exceeded() ) {
throw new \RuntimeException(
__( 'Monthly API budget exceeded. Generation halted.', 'prautoblogger' )
);
}
$target_count = absint( get_option( 'prautoblogger_daily_article_target', 1 ) );
// Step 1: Collect source data.
$this->broadcast_stage( __( 'Collecting sources from Reddit…', 'prautoblogger' ) );
$collector = new PRAutoBlogger_Source_Collector();
$collector->collect_from_all_sources();
// Step 2: Analyze collected data for patterns.
// Pass a generous target so the LLM returns enough diverse ideas for
// scoring to pick from — ask for 2× the daily target so dedup has room.
$this->broadcast_stage( __( 'Analyzing topics and scoring…', 'prautoblogger' ) );
$llm = new PRAutoBlogger_OpenRouter_Provider();
$analyzer = new PRAutoBlogger_Content_Analyzer( $llm, $cost_tracker );
$analysis = $analyzer->analyze_recent_data( max( $target_count * 2, 6 ) );
// Step 3: Score and deduplicate ideas.
$this->broadcast_stage( __( 'Selecting best topic…', 'prautoblogger' ) );
$scorer = new PRAutoBlogger_Idea_Scorer();
$scorer->set_skip_dedup( $this->skip_dedup );
$ideas = $scorer->score_and_rank( $analysis, $target_count );
$ideas = $this->orchestrate( $cost_tracker );
if ( empty( $ideas ) ) {
PRAutoBlogger_Logger::instance()->info( 'No viable article ideas found. Skipping generation.', 'pipeline' );
return [ 'generated' => 0, 'published' => 0, 'rejected' => 0, 'cost' => 0.0 ];
}
// Step 4 & 5: Generate content and run through editor for each idea.
$generator = new PRAutoBlogger_Content_Generator( $llm, $cost_tracker );
$editor = new PRAutoBlogger_Chief_Editor( $llm, $cost_tracker );
$publisher = new PRAutoBlogger_Publisher();
$auto_publish = in_array( get_option( 'prautoblogger_auto_publish', '0' ), [ '1', 'yes' ], true );
// Generate article 1 in this process — always fits within time limit.
$first_idea = array_shift( $ideas );
$worker = new PRAutoBlogger_Article_Worker( $cost_tracker );
$result = $worker->generate( $first_idea );
$generated = 0;
$published = 0;
$rejected = 0;
foreach ( $ideas as $idea ) {
// Re-check budget before each article — stop early if we hit the limit.
if ( $cost_tracker->is_budget_exceeded() ) {
PRAutoBlogger_Logger::instance()->warning( 'Budget limit reached during generation. Stopping early.', 'pipeline' );
break;
}
try {
$this->broadcast_stage( __( 'Generating article draft via AI…', 'prautoblogger' ) );
$content = $generator->generate( $idea );
$generated++;
$this->broadcast_stage( __( 'Running editorial pass…', 'prautoblogger' ) );
$review = $editor->review( $content, $idea );
if ( 'approved' === $review->get_verdict() || 'revised' === $review->get_verdict() ) {
// Null-coalesce: if editor returns 'revised' verdict but
// get_revised_content() is null (e.g. malformed LLM response),
// fall back to the original generated content.
$final_content = 'revised' === $review->get_verdict()
? ( $review->get_revised_content() ?? $content )
: $content;
$this->broadcast_stage( __( 'Saving and publishing…', 'prautoblogger' ) );
if ( $auto_publish ) {
$publisher->publish( $final_content, $idea, $review, $cost_tracker->get_run_id() );
$published++;
} else {
// Auto-publish disabled: send to review queue as draft.
$publisher->save_as_draft( $final_content, $idea, $review, $cost_tracker->get_run_id() );
}
} else {
$publisher->save_as_draft( $content, $idea, $review, $cost_tracker->get_run_id() );
$rejected++;
PRAutoBlogger_Logger::instance()->info( 'Article rejected by editor: ' . $idea->get_topic(), 'pipeline' );
}
} catch ( \Throwable $e ) {
PRAutoBlogger_Logger::instance()->error( sprintf( 'Article generation %s for "%s": %s', get_class( $e ), $idea->get_topic(), $e->getMessage() ), 'pipeline' );
// Continue with next idea — don't let one failure stop the batch.
}
// Queue remaining articles as chained cron events.
if ( ! empty( $ideas ) ) {
$this->queue_remaining( $ideas, $cost_tracker->get_run_id(), $result );
} else {
$this->log_summary( $result );
}
$total_cost = $cost_tracker->get_current_run_cost();
PRAutoBlogger_Logger::instance()->info(
sprintf( 'Pipeline complete: %d generated, %d published, %d rejected. Cost: $%.4f', $generated, $published, $rejected, $total_cost ),
'pipeline'
);
return [
'generated' => $generated,
'published' => $published,
'rejected' => $rejected,
'cost' => $total_cost,
];
return $result;
}
/**
* Update the generation status transient with the current pipeline stage.
* Process the next article from the persistent queue.
*
* Called at key checkpoints during the pipeline so the frontend status
* poller can show real progress instead of fake timers. No-ops if the
* transient doesn't exist (e.g., during daily cron runs).
* Called by WP-Cron for articles 2, 3, …N. Pops one idea, generates
* it, updates running totals, and chains the next job or finalizes.
*
* @param string $stage Human-readable stage description.
* Side effects: API calls, DB writes, post creation, cron scheduling.
*/
public function process_next_queued_article(): void {
$queue = get_option( self::QUEUE_KEY );
if ( ! is_array( $queue ) || empty( $queue['ideas'] ) ) {
$this->finish_queue();
return;
}
$idea_data = array_shift( $queue['ideas'] );
$idea = new PRAutoBlogger_Article_Idea( $idea_data );
$cost_tracker = new PRAutoBlogger_Cost_Tracker();
$cost_tracker->set_run_id( $queue['run_id'] );
if ( $cost_tracker->is_budget_exceeded() ) {
PRAutoBlogger_Logger::instance()->warning(
'Budget limit reached. Aborting remaining queued articles.',
'pipeline'
);
$this->finish_queue();
return;
}
$worker = new PRAutoBlogger_Article_Worker( $cost_tracker );
$article_result = $worker->generate( $idea );
// Merge into running totals.
$queue['results']['generated'] += $article_result['generated'];
$queue['results']['published'] += $article_result['published'];
$queue['results']['rejected'] += $article_result['rejected'];
$queue['results']['cost'] += $article_result['cost'];
if ( ! empty( $queue['ideas'] ) ) {
update_option( self::QUEUE_KEY, $queue, false );
$this->update_queue_status( $queue );
$this->schedule_next();
} else {
$this->write_final_status( $queue['results'] );
delete_option( self::QUEUE_KEY );
PRAutoBlogger_Generation_Lock::release();
$this->log_summary( $queue['results'] );
}
}
// ── Private helpers ─────────────────────────────────────────────────
/**
* Collect → Analyze → Score. Returns ranked ideas.
*
* @param PRAutoBlogger_Cost_Tracker $cost_tracker
* @return PRAutoBlogger_Article_Idea[]
*/
private function orchestrate( PRAutoBlogger_Cost_Tracker $cost_tracker ): array {
$target = absint( get_option( 'prautoblogger_daily_article_target', 1 ) );
$this->broadcast_stage( __( 'Collecting sources from Reddit…', 'prautoblogger' ) );
( new PRAutoBlogger_Source_Collector() )->collect_from_all_sources();
$this->broadcast_stage( __( 'Analyzing topics and scoring…', 'prautoblogger' ) );
$llm = new PRAutoBlogger_OpenRouter_Provider();
$analyzer = new PRAutoBlogger_Content_Analyzer( $llm, $cost_tracker );
$analysis = $analyzer->analyze_recent_data( max( $target * 2, 6 ) );
$this->broadcast_stage( __( 'Selecting best topic…', 'prautoblogger' ) );
$scorer = new PRAutoBlogger_Idea_Scorer();
$scorer->set_skip_dedup( $this->skip_dedup );
$ideas = $scorer->score_and_rank( $analysis, $target );
if ( empty( $ideas ) ) {
PRAutoBlogger_Logger::instance()->info(
'No viable article ideas found. Skipping generation.',
'pipeline'
);
}
return $ideas;
}
/**
* Store remaining ideas and schedule the next chained job.
*
* @param PRAutoBlogger_Article_Idea[] $ideas Remaining ideas (2..N).
* @param string $run_id Pipeline run UUID.
* @param array $first Results from article 1.
*/
private function queue_remaining( array $ideas, string $run_id, array $first ): void {
$serialized = array_map(
static fn( PRAutoBlogger_Article_Idea $i ) => $i->to_array(),
$ideas
);
$queue = [
'run_id' => $run_id,
'ideas' => $serialized,
'results' => $first,
];
update_option( self::QUEUE_KEY, $queue, false );
$this->update_queue_status( $queue );
$this->schedule_next();
PRAutoBlogger_Logger::instance()->info(
sprintf( 'Article 1 done. Queued %d more as chained jobs.', count( $serialized ) ),
'pipeline'
);
}
/** Schedule the next chained cron event and fire it immediately. */
private function schedule_next(): void {
if ( ! wp_next_scheduled( self::CRON_ACTION ) ) {
wp_schedule_single_event( time(), self::CRON_ACTION );
}
spawn_cron();
wp_remote_post(
site_url( 'wp-cron.php?doing_wp_cron=' . sprintf( '%.22F', microtime( true ) ) ),
[ 'timeout' => 0.01, 'blocking' => false, 'sslverify' => false ]
);
}
/** Update the frontend status transient with queue progress. */
private function update_queue_status( array $queue ): void {
$done = $queue['results']['generated'] + $queue['results']['rejected'];
$total = $done + count( $queue['ideas'] );
$current = get_transient( 'prautoblogger_generation_status' );
set_transient( 'prautoblogger_generation_status', [
'status' => 'running',
'stage' => sprintf( __( 'Generating article %1$d of %2$d…', 'prautoblogger' ), $done + 1, $total ),
'started' => is_array( $current ) ? ( $current['started'] ?? time() ) : time(),
], 600 );
}
/** Write the final "complete" status transient. */
private function write_final_status( array $r ): void {
set_transient( 'prautoblogger_generation_status', [
'status' => 'complete',
'generated' => $r['generated'],
'published' => $r['published'],
'rejected' => $r['rejected'],
'cost' => $r['cost'],
], 600 );
}
/** Clean up a stale or empty queue. */
private function finish_queue(): void {
$queue = get_option( self::QUEUE_KEY );
if ( is_array( $queue ) && isset( $queue['results'] ) ) {
$this->write_final_status( $queue['results'] );
}
delete_option( self::QUEUE_KEY );
PRAutoBlogger_Generation_Lock::release();
}
/** Log the pipeline summary. */
private function log_summary( array $r ): void {
PRAutoBlogger_Logger::instance()->info(
sprintf(
'Pipeline complete: %d generated, %d published, %d rejected. Cost: $%.4f',
$r['generated'], $r['published'], $r['rejected'], $r['cost']
),
'pipeline'
);
}
/** Update the generation status transient with the current stage. */
private function broadcast_stage( string $stage ): void {
$transient_key = 'prautoblogger_generation_status';
$current = get_transient( $transient_key );
$key = 'prautoblogger_generation_status';
$current = get_transient( $key );
if ( is_array( $current ) && 'running' === ( $current['status'] ?? '' ) ) {
$current['stage'] = $stage;
set_transient( $transient_key, $current, 600 );
set_transient( $key, $current, 600 );
}
}
}