Repository: oppia/oppia
File: core/jobs/batch_jobs/audit_threads_with_missing_suggestions_jobs.py (Python)

Code before:
            >> ndb_io.GetModels(
                feedback_models.GeneralFeedbackThreadModel.get_all(
                    include_deleted=False
                )
            )
            | 'Filter threads with missing suggestions'
            >> beam.Filter(
                lambda thread, suggestion_thread_ids: (
                    thread.has_suggestion
                    and thread.id not in suggestion_thread_ids
                ),
                beam.pvalue.AsList(valid_suggestion_thread_ids),
            )
        )


class AuditThreadsWithMissingSuggestionsJob(
    BaseThreadsWithMissingSuggestionsJob
):
    """Audit job reporting invalid feedback threads."""

    DATASTORE_UPDATES_ALLOWED = False

    def run(self) -> beam.PCollection[job_run_result.JobRunResult]:
        suggestion_thread_ids = self._get_suggestion_thread_ids()
        invalid_threads = self._get_invalid_threads(suggestion_thread_ids)

        invalid_thread_logs = invalid_threads | 'Log invalid threads' >> beam.Map(
            lambda model: job_run_result.JobRunResult.as_stdout(
                (
                    'GeneralFeedbackThreadModel marked as has_suggestion=True '
                    'but no GeneralSuggestionModel exists: '
                    f'id={model.id}'
                )
            )
        )

        invalid_thread_count = (
            invalid_threads
            | 'Count invalid threads'
            >> beam.combiners.Count.Globally().with_defaults(0)
            | 'Report invalid thread count'
            >> beam.Map(
                lambda count: job_run_result.JobRunResult.as_stdout(
                    f'invalid_feedback_thread_models_count: {count}'
                )
            )
        )

        return (
            invalid_thread_logs,

Review comment (suggestion):
By prefering PTransforms, this (and lines 130-131) can be changed to:

```python
invalid_threads = (
    self.pipeline
    | _GetSuggestionThreadIds()
    | _GetInvalidFeedbackThreads()
)
```

Code after:
        )

        feedback_threads = (
            self.pipeline
            | 'Get GeneralFeedbackThreadModels'
            >> ndb_io.GetModels(
                feedback_models.GeneralFeedbackThreadModel.get_all(
                    include_deleted=False
                )
            )
            | 'Keep threads with has_suggestion=True'
            >> beam.Filter(lambda thread: thread.has_suggestion)
        )

        feedback_id_to_model = (
            feedback_threads
            | 'Map feedback threads to (id, model)'
            >> beam.Map(lambda thread: (thread.id, thread))
        )

        suggestion_id_to_none = (
            suggestion_thread_ids
            | 'Map suggestion ids to None'
            >> beam.Map(lambda thread_id: (thread_id, None))
        )

        invalid_threads = (
            {
                'feedback': feedback_id_to_model,
                'suggestions': suggestion_id_to_none,
            }
            | 'CoGroup feedback threads with suggestion ids'
            >> beam.CoGroupByKey()
            | 'Select threads with missing suggestions'
            >> beam.FlatMap(
                lambda group: (
                    group[1]['feedback'] if not group[1]['suggestions'] else []
                )
            )
        )

        invalid_thread_logs = invalid_threads | 'Log invalid threads' >> beam.Map(
            lambda model: job_run_result.JobRunResult.as_stdout(
                (
                    'GeneralFeedbackThreadModel marked as has_suggestion=True '
                    'but no GeneralSuggestionModel exists: '
                    f'id={model.id}'
                )
            )
        )