Pull Requestメトリクスの計測・収集と可視化

import axios from "axios";
import "dotenv/config";
import fs from "fs";
import { Storage } from "@google-cloud/storage";
const bitbucketApi = axios.create({
  baseURL: "https://api.bitbucket.org/2.0",
  timeout: 10000,
  headers: { Authorization: `Bearer ${process.env.BITBUCKET_ACCESS_TOKEN}` },
});
const repoSlug = process.env.REPO || "";
const workspaceId = process.env.WORKSPACE || "";
const bucketName = process.env.BUCKET_NAME || "";
const getPRs = async (repoSlug: string, workspaceId: string) => {
  let hasMore = true;
  let page = 1;
  let allPRs: Object[] = [];
  while (hasMore) {
    const response = await bitbucketApi.get(
      `/repositories/${workspaceId}/${repoSlug}/pullrequests`,
      {
        params: {
          state: "MERGED",
          page: page,
        },
      }
    );
    allPRs = allPRs.concat(response.data.values);
    hasMore = Boolean(response.data.next);
    page++;
    console.log("page", page);
  }
  return allPRs;
};
const uploadFileToGCS = async (bucketName: string, filePath: string) => {
  const storage = new Storage();
  const bucket = storage.bucket(bucketName);
  const fileName = filePath.split("/").pop();
  try {
    await bucket.upload(filePath, {
      destination: fileName,
    });
    console.log(`Success: Uploaded ${filePath} to ${bucketName}/${fileName}`);
  } catch (error) {
    console.error("Error in GCS upload:", error);
  }
};
const main = async () => {
  try {
    const prs = await getPRs(repoSlug, workspaceId);
    const ndJsonContent = prs.map((obj) => JSON.stringify(obj)).join("\n");
    const filePath = `data/${repoSlug}-full.ndjson`;
    fs.writeFileSync(filePath, ndJsonContent);
    console.log("Success: write ndJson");
    await uploadFileToGCS(bucketName, filePath);
  } catch (error) {
    console.error("Error", error);
  }
};
main();

import json
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
pipeline_options = PipelineOptions(
    runner="DataflowRunner",
    project="your_project",
    temp_location="gs://your/backet-path/temp",
    region="asia-northeast1",
)
def flatten_json_array(json_str):
    json_arr = json.loads(json_str)
    return json_arr
def process_pr_data(pr_json_str):
    from datetime import datetime
    pr = json.loads(pr_json_str)
    created_on = datetime.fromisoformat(pr["created_on"].rstrip("Z"))
    updated_on = datetime.fromisoformat(pr["updated_on"].rstrip("Z"))
    duration_seconds = (updated_on - created_on).total_seconds()
    duration_days = duration_seconds / 86400
    return {
        "pr_id": pr["id"],
        "title": pr["title"],
        "author": pr["author"]["display_name"],
        "comment_count": pr["comment_count"],
        "created_on": created_on,
        "updated_on": updated_on,
        "duration_days": duration_days,
        "insertion_timestamp": datetime.utcnow(),
    }
schema = "pr_id:INTEGER, title:STRING, author:STRING, comment_count:INTEGER, created_on:TIMESTAMP, updated_on:TIMESTAMP, duration_days:FLOAT, insertion_timestamp:TIMESTAMP"
def run():
    with beam.Pipeline(options=pipeline_options) as pipeline:
        prs = (
            pipeline
            | "ReadFromGCS"
            >> beam.io.ReadFromText("gs://backet-path/path.ndjson")
            | "ProcessPRData" >> beam.Map(process_pr_data)
        )
        prs | "WriteToBigQuery" >> beam.io.WriteToBigQuery(
            "project:dataset-path.table-path",
            schema=schema,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        )
if __name__ == "__main__":
    run()