import * as fs from 'fs'; import * as path from 'path'; import * as request from 'request'; import * as Twitter from 'twitter'; import TwitterTypes from 'twitter-d'; import { getLogger } from './loggers'; import QQBot, { Message, MessageChain } from './mirai'; import { BigNumOps } from './utils'; import Webshot from './webshot'; interface IWorkerOption { lock: ILock; lockfile: string; bot: QQBot; workInterval: number; webshotDelay: number; consumer_key: string; consumer_secret: string; access_token_key: string; access_token_secret: string; private_csrf_token: string; private_auth_token: string; mode: number; } export class ScreenNameNormalizer { // tslint:disable-next-line: variable-name public static _queryUser: (username: string) => Promise; public static permaFeeds = {}; public static savePermaFeedForUser(user: FullUser) { this.permaFeeds[`https://twitter.com/${user.screen_name}`] = `https://twitter.com/i/user/${user.id_str}`; } public static normalize = (username: string) => username.toLowerCase().replace(/^@/, ''); public static async normalizeLive(username: string) { if (this._queryUser) { return await this._queryUser(username) .catch((err: {code: number, message: string}[]) => { if (err[0].code !== 50) { logger.warn(`error looking up user: ${err[0].message}`); return username; } return null; }); } return this.normalize(username); } } export let sendAllFleets = (username: string, receiver: IChat): void => { throw Error(); }; const logger = getLogger('twitter'); const maxTrials = 3; const uploadTimeout = 10000; const retryInterval = 1500; const ordinal = (n: number) => { switch ((Math.trunc(n / 10) % 10 === 1) ? 0 : n % 10) { case 1: return `${n}st`; case 2: return `${n}nd`; case 3: return `${n}rd`; default: return `${n}th`; } }; const retryOnError = ( doWork: () => Promise, onRetry: (error, count: number, terminate: (defaultValue: U) => void) => void ) => new Promise(resolve => { const retry = (reason, count: number) => { setTimeout(() => { let terminate = false; onRetry(reason, count, defaultValue => { terminate = true; resolve(defaultValue); }); if (!terminate) doWork().then(resolve).catch(error => retry(error, count + 1)); }, retryInterval); }; doWork().then(resolve).catch(error => retry(error, 1)); }); export type FullUser = TwitterTypes.FullUser; export type MediaEntity = TwitterTypes.MediaEntity; type TwitterMod = { -readonly [K in keyof Twitter]: Twitter[K]; } & { options?: any; }; interface IFleet { created_at: string; deleted_at: string; expiration: string; fleet_id: string; fleet_thread_id: string; media_bounding_boxes: [{ anchor_point_x: number; anchor_point_y: number; width: number; height: number; rotation: number; entity: { type: string; value: any; } }]; media_entity: MediaEntity; media_key: { media_category: 'TWEET_IMAGE' | 'TWEET_VIDEO'; media_id: number; media_id_str: string; }; mentions: any; mentions_str: any; read: boolean; text: string; user_id: number; user_id_str: string; } export type Fleet = IFleet; export type Fleets = IFleet[]; interface IFleetFeed { fleet_threads: {fleets: Fleets}[]; } export default class { private client: Twitter; private privateClient: TwitterMod; private lock: ILock; private lockfile: string; private workInterval: number; private bot: QQBot; private webshotDelay: number; private webshot: Webshot; private mode: number; constructor(opt: IWorkerOption) { this.client = new Twitter({ consumer_key: opt.consumer_key, consumer_secret: opt.consumer_secret, access_token_key: opt.access_token_key, access_token_secret: opt.access_token_secret, }); this.privateClient = new Twitter({ bearer_token: 'AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA', } as any); this.privateClient.request = request.defaults({ headers: { ...this.privateClient.options.request_options.headers, 'Content-Type': 'application/x-www-form-urlencoded', Cookie: `auth_token=${opt.private_auth_token}; ct0=${opt.private_csrf_token};`, 'X-CSRF-Token': opt.private_csrf_token, }, }); this.lockfile = opt.lockfile; this.lock = opt.lock; this.workInterval = opt.workInterval; this.bot = opt.bot; this.mode = opt.mode; ScreenNameNormalizer._queryUser = this.queryUser; sendAllFleets = (username, receiver) => { this.client.get('users/show', {screen_name: username}) .then((user: FullUser) => { const feed = `https://twitter.com/${user.screen_name}`; return this.getFleets(user.id_str) .catch(error => { logger.error(`unhandled error while fetching fleets for ${feed}: ${JSON.stringify(error)}`); this.bot.sendTo(receiver, `获取 Fleets 时出现错误:${error}`); }) .then((fleetFeed: IFleetFeed) => { if (!fleetFeed || fleetFeed.fleet_threads.length === 0) { this.bot.sendTo(receiver, `当前用户(@${user.screen_name})没有可用的 Fleets。`); return; } this.workOnFleets(user, fleetFeed.fleet_threads[0].fleets, this.sendFleets(`thread ${feed}`, receiver)); }); }) .catch((err: {code: number, message: string}[]) => { if (err[0].code !== 50) { logger.warn(`error looking up user: ${err[0].message}, unable to fetch fleets`); } this.bot.sendTo(receiver, `找不到用户 ${username.replace(/^@?(.*)$/, '@$1')}。`); }); }; } public launch = () => { this.webshot = new Webshot( this.mode, () => setTimeout(this.work, this.workInterval * 1000) ); } public queryUser = (username: string) => this.client.get('users/show', {screen_name: username}) .then((user: FullUser) => { ScreenNameNormalizer.savePermaFeedForUser(user); return user.screen_name; }) private workOnFleets = ( user: FullUser, fleets: Fleets, sendFleets: (msg: MessageChain, text: string) => void ) => { const uploader = ( message: ReturnType, lastResort: (...args) => ReturnType ) => { let timeout = uploadTimeout; return retryOnError(() => this.bot.uploadPic(message, timeout).then(() => message), (_, count, terminate: (defaultValue: ReturnType) => void) => { if (count <= maxTrials) { timeout *= (count + 2) / (count + 1); logger.warn(`retry uploading for the ${ordinal(count)} time...`); } else { logger.warn(`${count - 1} consecutive failures while uploading, trying plain text instead...`); terminate(lastResort()); } }); }; return this.webshot(user, fleets, uploader, sendFleets, this.webshotDelay); } private sendFleets = (source?: string, ...to: IChat[]) => (msg: MessageChain, text: string) => { to.forEach(subscriber => { logger.info(`pushing data${source ? ` of ${source}` : ''} to ${JSON.stringify(subscriber)}`); retryOnError( () => this.bot.sendTo(subscriber, msg), (_, count, terminate: (doNothing: Promise) => void) => { if (count <= maxTrials) { logger.warn(`retry sending to ${subscriber.chatID} for the ${ordinal(count)} time...`); } else { logger.warn(`${count - 1} consecutive failures while sending` + 'message chain, trying plain text instead...'); terminate(this.bot.sendTo(subscriber, text)); } }); }); } private getFleets = (userID: string) => new Promise((resolve, reject) => { const endpoint = `https://api.twitter.com/fleets/v1/user_fleets?user_id=${userID}`; this.privateClient.get(endpoint, (error, fleetFeed: IFleetFeed, _) => { if (error) reject(error); else resolve(fleetFeed); }); }) public work = () => { const lock = this.lock; if (this.workInterval < 1) this.workInterval = 1; if (lock.feed.length === 0) { setTimeout(() => { this.work(); }, this.workInterval * 1000); return; } if (lock.workon >= lock.feed.length) lock.workon = 0; if (!lock.threads[lock.feed[lock.workon]] || !lock.threads[lock.feed[lock.workon]].subscribers || lock.threads[lock.feed[lock.workon]].subscribers.length === 0) { logger.warn(`nobody subscribes thread ${lock.feed[lock.workon]}, removing from feed`); delete lock.threads[lock.feed[lock.workon]]; lock.feed.splice(lock.workon, 1); fs.writeFileSync(path.resolve(this.lockfile), JSON.stringify(lock)); this.work(); return; } const currentFeed = lock.feed[lock.workon]; logger.debug(`pulling feed ${currentFeed}`); let user: FullUser; let match = currentFeed.match(/https:\/\/twitter.com\/([^\/]+)/); if (match) match = lock.threads[currentFeed].permaFeed.match(/https:\/\/twitter.com\/i\/user\/([^\/]+)/); if (!match) { logger.error(`cannot get endpoint for feed ${currentFeed}`); return; } this.client.get('users/show', {user_id: match[1]}) .then((fullUser: FullUser) => { user = fullUser; return this.getFleets(match[1]); }) .catch(error => { logger.error(`unhandled error on fetching fleets for ${currentFeed}: ${JSON.stringify(error)}`); }) .then((fleetFeed: IFleetFeed) => { logger.debug(`private api returned ${JSON.stringify(fleetFeed)} for feed ${currentFeed}`); logger.debug(`api returned ${JSON.stringify(user)} for owner of feed ${currentFeed}`); const currentThread = lock.threads[currentFeed]; const updateDate = () => currentThread.updatedAt = new Date().toString(); if (!fleetFeed || fleetFeed.fleet_threads.length === 0) { updateDate(); return; } let fleets = fleetFeed.fleet_threads[0].fleets; const bottomOfFeed = fleets.slice(-1)[0].fleet_id.substring(3); const updateOffset = () => currentThread.offset = bottomOfFeed; if (currentThread.offset === '-1') { updateOffset(); return; } if (currentThread.offset !== '0') { const readCount = fleets.findIndex(fleet => Number(BigNumOps.plus(fleet.fleet_id.substring(3), `-${currentThread.offset}`)) > 0); if (readCount === -1) return; fleets = fleets.slice(readCount); } return this.workOnFleets(user, fleets, this.sendFleets(`thread ${currentFeed}`, ...currentThread.subscribers)) .then(updateDate).then(updateOffset); }) .then(() => { lock.workon++; let timeout = this.workInterval * 1000 / lock.feed.length; if (timeout < 1000) timeout = 1000; fs.writeFileSync(path.resolve(this.lockfile), JSON.stringify(lock)); setTimeout(() => { this.work(); }, timeout); }); } }