feat: add oss client and hypercrx task (#772)

Signed-off-by: Frankzhaopku <syzhao1988@126.com>

package.json

@@ -37,6 +37,7 @@
     "lodash": "^4.17.21",
     "neo4j-driver": "^4.4.1",
     "node-cron": "^3.0.0",
+    "p-wait-for": "^3.1.0",
     "parse-neo4j": "^0.6.11",
     "pope": "^3.0.0",
     "request": "^2.88.2",

src/config.ts

@@ -1,5 +1,7 @@
 import { merge } from "lodash";
 
+let inited = false;
+
 let config = {
   general: {
     owner: 'X-lab2017',
@@ -29,8 +31,13 @@ let config = {
   }
 };
 
-import('./local_config').then(localConfig => {
-  config = merge(config, localConfig.default);
-}).catch(() => {});
-
-export default () => config;
+export default async () => {
+  if (!inited) {
+    try {
+      await import('./local_config').then(localConfig => {
+        config = merge(config, localConfig.default);
+      });
+    } catch {}
+  }
+  return config;
+}

src/cron/tasks/hacking_force_month.ts

@@ -7,7 +7,7 @@ import { getClient } from '../../oss/ali';
 const task: Task = {
   cron: '0 0 5 * *',    // runs on the 5th day of every month at 00:00
   enable: true,
-  immediate: true,
+  immediate: false,
   callback: async () => {
     const chineseUserIds = getGitHubData([':regions/China']).githubUsers;
     const q = `MATCH (u:User) WHERE u.id IN [${chineseUserIds.join(',')}] RETURN u;`;
@@ -32,7 +32,7 @@ const task: Task = {
         });
       });
     });
-    const client = getClient();
+    const client = await getClient();
     await client.put('/hacking_force/total.json', Buffer.from(JSON.stringify(result)));
     console.log('Run hacking force total task done.');
   }

src/cron/tasks/hypercrx_actor.ts

@@ -0,0 +1,50 @@
+import { existsSync, mkdirSync, writeFileSync } from 'fs';
+import { Task } from '..';
+import { getClient as getNeo4jClient, parseRecord } from '../../db/neo4j';
+import { forEveryMonth } from '../../metrics/basic';
+
+const task: Task = {
+  cron: '0 0 7 * *',    // runs on the 5th day of every month at 00:00
+  enable: true,
+  immediate: false,
+  callback: async () => {
+    const neo4jClient = await getNeo4jClient();
+    const session = neo4jClient.session();
+    const now = new Date();
+    const lastMonth = new Date(now.getTime() - 30 * 24 * 60 * 60);
+    const year = lastMonth.getFullYear(), month = lastMonth.getMonth() + 1;
+    const q = `MATCH (u:User) WHERE u.activity_${year}${month} IS NOT NULL RETURN u;`;
+    const result = session.run(q);
+    let count = 0;
+    result.subscribe({
+      onNext: async r => {
+        const user = parseRecord(r);
+        const userInfo = {
+          login: user.login,
+          id: user.id,
+          activity: {},
+          influence: {},
+        };
+        forEveryMonth(2015, 1, year, month, (y, m) => {
+          userInfo.activity[`${y}-${m}`] = user[`activity_${y}${m}`] ?? 0;
+          userInfo.influence[`${y}-${m}`] = user[`open_rank_${y}${m}`] ?? 0;
+        });
+        const dir = `./local_files/hypercrx_actor/${user.login.charAt(0).toLowerCase()}`;
+        if (!existsSync(dir)) {
+          mkdirSync(dir);
+        }
+        writeFileSync(`${dir}/${user.login}.json`, JSON.stringify(userInfo));
+        count++;
+        if (count % 10000 === 0) console.log(`Finish write ${count} files.`);
+      },
+      onCompleted: () => {
+        session.close().then(() => {
+          console.log(`Process ${count} users done.`);
+        });
+      },
+      onError: console.log,
+    });
+  }
+};
+
+module.exports = task;

src/cron/tasks/hypercrx_repo.ts

@@ -0,0 +1,53 @@
+import { writeFileSync, existsSync, mkdirSync } from 'fs';
+import { Task } from '..';
+import { getClient as getNeo4jClient, parseRecord } from '../../db/neo4j';
+import { forEveryMonth } from '../../metrics/basic';
+
+const task: Task = {
+  cron: '0 0 6 * *',    // runs on the 5th day of every month at 00:00
+  enable: true,
+  immediate: false,
+  callback: async () => {
+    const neo4jClient = await getNeo4jClient();
+    const session = neo4jClient.session();
+    const now = new Date();
+    const lastMonth = new Date(now.getTime() - 30 * 24 * 60 * 60);
+    const year = lastMonth.getFullYear(), month = lastMonth.getMonth() + 1;
+    const q = `MATCH (r:Repo) WHERE r.activity_${year}${month} IS NOT NULL RETURN r;`;
+    const result = session.run(q);
+    let count = 0;
+    result.subscribe({
+      onNext: async r => {
+        const repo = parseRecord(r);
+        const repoInfo = {
+          id: repo.id,
+          name: repo.name,
+          org: repo.org_login,
+          org_id: repo.org_id,
+          activity: {},
+          influence: {},
+        };
+        forEveryMonth(2015, 1, year, month, (y, m) => {
+          repoInfo.activity[`${y}-${m}`] = repo[`activity_${y}${m}`] ?? 0;
+          repoInfo.influence[`${y}-${m}`] = repo[`open_rank_${y}${m}`] ?? 0;
+        });
+        const [owner, name] = repoInfo.name.split('/');
+        const dir = `./local_files/hypercrx_repo/${owner}`;
+        if (!existsSync(dir)) {
+          mkdirSync(dir);
+        }
+        writeFileSync(`${dir}/${name}.json`, JSON.stringify(repoInfo));
+        count++;
+        if (count % 10000 === 0) console.log(`Finish write ${count} files.`);
+      },
+      onCompleted: () => {
+        session.close().then(() => {
+          console.log(`Process ${count} repos done.`);
+        });
+      },
+      onError: console.log,
+    });
+  }
+};
+
+module.exports = task;

src/db/clickhouse.ts

@@ -3,13 +3,13 @@ import getConfig from '../config';
 
 let client;
 
-function getClient() {
+async function getClient() {
   if (client) return client;
-  client = new ClickHouse(getConfig().db.clickhouse);
+  client = new ClickHouse((await getConfig()).db.clickhouse);
   return client;
 }
 
 export async function query<T>(q: string): Promise<T> {
-    const client = getClient();
+    const client = await getClient();
     return (await client.querying(q) as any).data;
 }

src/db/neo4j.ts

@@ -4,15 +4,19 @@ import getConfig from '../config';
 
 let driver: any;
 
-function getClient() {
+export async function getClient() {
   if (driver) return driver;
-  driver = neo4j.driver(getConfig().db.neo4j.host);
+  driver = neo4j.driver((await getConfig()).db.neo4j.host);
   return driver;
 }
 
-export async function query<T = any>(query: string, params?: any) {
-  const session = getClient().session();
+export async function query<T = any>(query: string, params?: any): Promise<T[]> {
+  const session = (await getClient()).session();
   const r = await session.run(query, params);
   await session.close();
-  return parser.parse(r) as T;
+  return parser.parse(r) as T[];
+}
+
+export function parseRecord<T = any>(record: any): T {
+  return parser.parseRecord(record)._fields[0];
 }

src/oss/ali.ts

@@ -3,9 +3,9 @@ import getConfig from '../config';
 
 let client: OSS;
 
-export function getClient(): OSS {
+export async function getClient(): Promise<OSS> {
   if (client) return client;
-  const config = getConfig();
+  const config = await getConfig();
   client = new OSS(config.oss.ali);
   return client;
 };

src/utils.ts

@@ -1,5 +1,6 @@
 import { existsSync, readFileSync } from "fs";
 import { load } from 'js-yaml';
+import pWaitFor from 'p-wait-for';
 
 export function readFileAsObj(path: string) {
   if (!existsSync(path)) {
@@ -25,3 +26,16 @@ export function readFileAsObj(path: string) {
   }
   return null;
 }
+
+export async function waitFor(mill: number): Promise<void> {
+  return new Promise(resolve => {
+    setTimeout(() => {
+      resolve();
+    }, mill);
+  });
+}
+
+export async function waitUntil(func: () => boolean, options?: object): Promise<void> {
+  if (func()) return;
+  return pWaitFor(func, Object.assign({ interval: 1000 }, options));
+}