import * as crypto from 'crypto'; import * as fs from 'fs'; import * as http from 'http'; import * as path from 'path'; import { parse as parseUrl } from 'url'; import { promisify } from 'util'; import { IgApiClient, IgClientError, IgCookieNotFoundError, IgExactUserNotFoundError, IgLoginTwoFactorRequiredError, IgLoginRequiredError, IgNetworkError, ReelsMediaFeedResponseItem, UserFeedResponseUser } from 'instagram-private-api'; import { RequestError } from 'request-promise/errors'; import { SocksProxyAgent } from 'socks-proxy-agent'; import { relativeDate } from './datetime'; import { getLogger } from './loggers'; import QQBot from './koishi'; import { Arr, BigNumOps, chainPromises } from './utils'; import Webshot from './webshot'; const parseLink = (link: string): {userName?: string, storyId?: string} => { let match = /instagram\.com\/stories\/([^\/?#]+)\/(\d+)/.exec(link); if (match) return {userName: ScreenNameNormalizer.normalize(match[1]).split(':')[0], storyId: match[2]}; match = /instagram\.com\/([^\/?#]+)/.exec(link) || /^([^\/?#]+)$/.exec(link); if (match) return {userName: ScreenNameNormalizer.normalize(match[1]).split(':')[0]}; return; }; const linkBuilder = (config: ReturnType): string => { if (!config.userName) return; if (!config.storyId) return `https://www.instagram.com/${config.userName}/`; return `https://www.instagram.com/stories/${config.userName}/${config.storyId}/`; }; export {linkBuilder, parseLink}; interface IWorkerOption { sessionLockfile: string; credentials: [string, string]; codeServicePort: number; proxyUrl: string; lock: ILock; lockfile: string; webshotCookiesLockfile: string; bot: QQBot; inactiveHours: string[]; workInterval: number; webshotDelay: number; mode: number; wsUrl: string; } export class SessionManager { private ig: IgApiClient; private username: string; private password: string; private lockfile: string; private codeServicePort: number; constructor(client: IgApiClient, file: string, credentials: [string, string], codeServicePort: number) { this.ig = client; this.lockfile = file; [this.username, this.password] = credentials; this.codeServicePort = codeServicePort; } public init = () => { this.ig.state.generateDevice(this.username); this.ig.request.end$.subscribe(() => { this.save(); }); const filePath = path.resolve(this.lockfile); if (fs.existsSync(filePath)) { try { const serialized = JSON.parse(fs.readFileSync(filePath, 'utf8')) as {[key: string]: any}; return this.ig.state.deserialize(serialized).then(() => { logger.info(`successfully loaded client session cookies for user ${this.username}`); }); } catch (err) { logger.error(`failed to load client session cookies from file ${this.lockfile}: `, err); return Promise.resolve(); } } else { return this.login().catch((err: IgClientError) => { logger.error(`error while trying to log in as user ${this.username}, error: ${err}`); logger.warn('attempting to retry after 1 minute...'); if (fs.existsSync(filePath)) fs.unlinkSync(filePath); promisify(setTimeout)(60000).then(this.init); }); } }; public handle2FA = (submitter: (code: string) => Promise) => new Promise((resolve, reject) => { const token = crypto.randomBytes(20).toString('hex'); logger.info('please submit the code with a one-time token from your browser with this path:'); logger.info(`/confirm-2fa?code=&token=${token}`); let working; const server = http.createServer((req, res) => { const {pathname, query} = parseUrl(req.url, true); if (!working && pathname === '/confirm-2fa' && query.token === token && typeof(query.code) === 'string' && /^\d{6}$/.test(query.code)) { const code = query.code; logger.debug(`received code: ${code}`); working = true; submitter(code) .then(response => { res.write('OK'); res.end(); server.close(() => resolve(response)); }) .catch(err => { res.write('Error'); res.end(); reject(err); }) .finally(() => { working = false; }); } }); server.listen(this.codeServicePort); }); public login = () => this.ig.simulate.preLoginFlow() .then(() => this.ig.account.login(this.username, this.password)) .catch((err: IgClientError) => { if (err instanceof IgLoginTwoFactorRequiredError) { const {two_factor_identifier, totp_two_factor_on} = err.response.body.two_factor_info; logger.debug(`2FA info: ${JSON.stringify(err.response.body.two_factor_info)}`); logger.info(`login is requesting two-factor authentication via ${totp_two_factor_on ? 'TOTP' : 'SMS'}`); return this.handle2FA(code => this.ig.account.twoFactorLogin({ username: this.username, verificationCode: code, twoFactorIdentifier: two_factor_identifier, verificationMethod: totp_two_factor_on ? '0' : '1', })); } throw err; }) .then(user => new Promise(resolve => { logger.info(`successfully logged in as ${this.username}`); process.nextTick(() => resolve(this.ig.simulate.postLoginFlow().then(() => user))); })); public save = () => this.ig.state.serialize() .then((serialized: {[key: string]: any}) => { delete serialized.constants; return fs.writeFileSync(path.resolve(this.lockfile), JSON.stringify(serialized, null, 2), 'utf-8'); }); } export class ScreenNameNormalizer { // tslint:disable-next-line: variable-name public static _queryUser: (username: string) => Promise; public static normalize = (username: string) => `${username.toLowerCase().replace(/^@/, '')}:`; public static async normalizeLive(username: string) { if (this._queryUser) { return await this._queryUser(username) .catch((err: IgClientError) => { if (!(err instanceof IgExactUserNotFoundError)) { logger.warn(`error looking up user: ${err.message}`); return `${username}:`; } return null; }); } return this.normalize(username); } } export let sendTimeline = (username: string, receiver: IChat): void => { throw Error(); } export let sendStory = (username: string, storyId: string, receiver: IChat): void => { throw Error(); } export let sendAllStories = (username: string, receiver: IChat, startIndex: number, count: number): void => { throw Error(); }; export type MediaItem = ReelsMediaFeedResponseItem; type CachedMediaItem = {pk: string, msgs: string, text: string, author: string, original: MediaItem}; const logger = getLogger('instagram'); const maxTrials = 3; 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 default class { private client: IgApiClient; private lock: ILock; private lockfile: string; private inactiveHours: string[]; private workInterval: number; private bot: QQBot; private webshotDelay: number; private webshot: Webshot; private mode: number; private wsUrl: string; public session: SessionManager; constructor(opt: IWorkerOption) { this.client = new IgApiClient(); if (opt.proxyUrl) { try { const url = new URL(opt.proxyUrl); if (!/^socks(?:4a?|5h?)?:$/.test(url.protocol)) throw Error(); if (!url.port) url.port = '1080'; this.client.request.defaults.agent = new SocksProxyAgent({ hostname: url.hostname, port: url.port, userId: url.username, password: url.password, }); } catch (e) { logger.warn(`invalid socks proxy url: ${opt.proxyUrl}, ignoring`); } } this.session = new SessionManager(this.client, opt.sessionLockfile, opt.credentials, opt.codeServicePort); this.lockfile = opt.lockfile; this.lock = opt.lock; this.inactiveHours = opt.inactiveHours; this.workInterval = opt.workInterval; this.bot = opt.bot; this.webshotDelay = opt.webshotDelay; this.mode = opt.mode; this.wsUrl = opt.wsUrl; const workNow = (config: { rawUserName: string, action: (userName: string) => void, retryAction: () => void, reply: (msg: string) => void }) => { const {action, retryAction, reply, rawUserName} = config; return this.queryUser(rawUserName) .then(userNameId => { const [userName, userId] = userNameId.split(':'); if (Date.now() - this.cache[userName]?.updated?.getTime() > this.workInterval * 1000 && Object.keys(this.cache[userName].stories).length > 0) { return userName; } return this.client.feed.reelsMedia({userIds: [userId]}).items() .then(storyItems => Promise.all(storyItems .filter(item => !(item.pk in this.cache[userName].stories)) .map(item => this.webshot( [{...item, user: this.cache[userName].user}], (msgs: string, text: string, author: string) => this.cache[userName].stories[item.pk] = {pk: item.pk, msgs, text, author, original: item}, this.webshotDelay )) ).then(() => userName).finally(() => this.cache[userName].updated = new Date())); }) .then(action) .catch((error: IgClientError & Partial) => { if (error instanceof IgExactUserNotFoundError) { reply(`找不到用户 ${rawUserName.replace(/^@?(.*)$/, '@$1')}。`); } if (error instanceof IgNetworkError) { if ((error.cause as Error).message === "Unexpected '<'") { logger.warn('login required, logging in again...'); return this.session.login().then(retryAction); } logger.warn(`error while fetching stories for ${rawUserName}: ${JSON.stringify(error.cause)}`); reply(`获取 Stories 时出现错误:原因: ${error.cause}`); } else if (error instanceof IgLoginRequiredError || error instanceof IgCookieNotFoundError) { logger.warn('login required, logging in again...'); reply('等待登陆中,稍后会处理请求,请稍候……'); this.session.login().then(retryAction); } else { logger.error(`unhandled error while fetching stories for ${rawUserName}: ${error}`); reply(`获取 Stories 时发生未知错误: ${error}`); } }); }; ScreenNameNormalizer._queryUser = this.queryUser; sendTimeline = (rawUserName, receiver) => { const reply = msg => this.bot.sendTo(receiver, msg); workNow({ rawUserName, action: userName => { const storyItems = Object.values(this.cache[userName].stories) .sort((i1, i2) => -BigNumOps.compare(i2.pk, i1.pk)); // ascending! if (storyItems.length === 0) return reply(`当前用户 (@${userName}) 没有可用的 Instagram 限时动态。`); return reply('#. 编号:发送时间\n' + storyItems.map(({original}, index) => `\n${index + 1}. ${original.pk}: ${relativeDate(original.taken_at * 1000)}`).join('')) .then(() => reply(`请使用 /igstory_view ${userName} skip=<#-1> count=1 或 /igstory_view https://www.instagram.com/stories/${userName}/<编号>/ 查看指定的限时动态。`)); }, reply, retryAction: () => sendTimeline(rawUserName, receiver), }); } sendStory = (rawUserName, storyId, receiver) => { const reply = msg => this.bot.sendTo(receiver, msg); const sender = this.sendStories(`instagram stories for ${rawUserName}`, receiver); workNow({ rawUserName, action: userName => { if (!(storyId in this.cache[userName].stories)) return reply('此动态不存在或已过期。'); return this.workOnMedia([this.cache[userName].stories[storyId]], sender); }, reply, retryAction: () => sendStory(rawUserName, storyId, receiver), }); } sendAllStories = (rawUserName, receiver, startIndex = 0, count = 10) => { const reply = msg => this.bot.sendTo(receiver, msg); if (startIndex < 0) return reply('跳过数量参数值应为非负整数。'); if (count < 1) return reply('最大查看数量参数值应为正整数。'); const sender = this.sendStories(`instagram stories for ${rawUserName}`, receiver); workNow({ rawUserName, action: userName => { const storyItems = Object.values(this.cache[userName].stories) .sort((i1, i2) => -BigNumOps.compare(i2.pk, i1.pk)); // ascending! if (storyItems.length === 0) return reply(`当前用户 (@${userName}) 没有可用的 Instagram 限时动态。`); if (startIndex + 1 > storyItems.length) return reply('跳过数量到达或超过当前用户可用的限时动态数量。'); const endIndex = Math.min(storyItems.length, startIndex + count); const sendRangeText = `${startIndex + 1}${endIndex - startIndex > 1 ? `-${endIndex}` : ''}`; return this.workOnMedia(storyItems.slice(startIndex, endIndex), sender) .then(() => reply(`已显示当前用户 ${storyItems.length} 条可用限时动态中的第 ${sendRangeText} 条。`)); }, reply, retryAction: () => sendAllStories(rawUserName, receiver, startIndex, count) }); }; } public launch = () => { this.webshot = new Webshot( this.wsUrl, this.mode, () => { setTimeout(this.workForAll, this.workInterval * 1000 / this.lock.feed.length); setTimeout(() => { this.work(); setInterval(() => { this.pullOrders = Arr.shuffle(this.pullOrders); }, 21600000); setInterval(this.workForAll, this.workInterval * 1000); }, this.workInterval * 1200 / this.lock.feed.length); } ); }; public queryUser = (rawUserName: string) => { const username = ScreenNameNormalizer.normalize(rawUserName).split(':')[0]; if (username in this.cache) { return Promise.resolve(`${username}:${this.cache[username].user.pk}`); } return this.client.user.searchExact(username) .catch((error: IgClientError) => { if (error instanceof IgLoginRequiredError || error instanceof IgCookieNotFoundError) { logger.warn('login required, logging in again...'); return this.session.login().then(() => this.client.user.searchExact(username)); } else throw error; }) .then(user => { logger.info(`initialized cache item for user ${user.full_name} (@${username})`); this.cache[user.username] = {user, stories: {}, pullOrder: 0}; return `${user.username}:${user.pk}`; }); }; private workOnMedia = ( mediaItems: CachedMediaItem[], sendMedia: (msg: string, text: string, author: string) => void ) => Promise.resolve(mediaItems.forEach(({msgs, text, author}) => sendMedia(msgs, text, author))); private sendStories = (source?: string, ...to: IChat[]) => (msg: string, text: string, author: 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, author + text, true)); } }); }); }; private cache: { [userName: string]: { user: UserFeedResponseUser & ReelsMediaFeedResponseItem['user'], stories: {[storyId: string]: CachedMediaItem}, pullOrder: number, // one-based; -1: subscribed, awaiting shuffle; 0: not subscribed updated?: Date, }, } = {}; private get pullOrders() { const arr: number[] = []; Object.values(this.cache).forEach(item => { if (item.pullOrder > 0) arr[item.pullOrder - 1] = item.user.pk; }); return arr; }; private set pullOrders(arr: number[]) { Object.values(this.cache).forEach(item => { item.pullOrder = arr.indexOf(item.user.pk) + 1; }); } private workForAll = () => { if (this.isInactiveTime) return; logger.debug(`current cache: ${JSON.stringify(this.cache)}`); chainPromises(Object.entries(this.lock.threads).map(([feed, thread]) => { const id = thread.id; const userName = parseLink(feed).userName; logger.debug(`preparing to add user @${userName} to next pull task...`); return (map: {[key: number]: UserFeedResponseUser} = {}) => { if (userName in this.cache) { const item = this.cache[userName]; if (item.pullOrder === 0) item.pullOrder = -1; return Promise.resolve(Object.assign(map, {[id]: item.user})); } return promisify(setTimeout)((Math.random() * 2 + 1) * 5000).then(() => this.client.user.info(id).then(user => { logger.info(`initialized cache item for user ${user.full_name} (@${userName})`); this.cache[userName] = {user, stories: {}, pullOrder: -1}; return Object.assign(map, {[id]: user}); }) ); }; })) .then(idToUserMap => { const userIdCache = Object.values(this.cache).some(item => item.pullOrder < 0) ? this.pullOrders = Arr.shuffle(Object.keys(idToUserMap)).map(Number) : this.pullOrders; return chainPromises( Arr.chunk(userIdCache, 20).map(userIds => () => { const itemToUserName = (item: MediaItem) => idToUserMap[item.user.pk].username; logger.info(`pulling stories from users:${userIds.map(id => ` @${idToUserMap[id].username}`)}`); return this.client.feed.reelsMedia({userIds}).items() .then(storyItems => Promise.all(storyItems .filter(item => !(item.pk in this.cache[itemToUserName(item)].stories)) .map(item => this.webshot( [{...item, user: this.cache[itemToUserName(item)].user}], (msgs: string, text: string, author: string) => this.cache[itemToUserName(item)].stories[item.pk] = {pk: item.pk, msgs, text, author, original: item}, this.webshotDelay )) )) .finally(() => Object.values(this.lock.threads).forEach(thread => { if (userIds.includes(thread.id)) { thread.updatedAt = (this.cache[idToUserMap[thread.id].username].updated = new Date()).toString(); } })) as unknown as Promise; }), (lp1, lp2) => () => lp1().then(() => promisify(setTimeout)(this.workInterval * 1000 / this.lock.feed.length).then(lp2)) ); }) .catch((error: IgClientError & Partial) => { if (error instanceof IgNetworkError) { if ((error.cause as Error).message === "Unexpected '<'") { logger.warn('login required, logging in again...'); return this.session.login().then(this.workForAll); } logger.warn(`error while fetching stories for all: ${JSON.stringify(error.cause)}`); } else if (error instanceof IgLoginRequiredError || error instanceof IgCookieNotFoundError) { logger.warn('login required, logging in again...'); this.session.login().then(this.workForAll); } else { logger.error(`unhandled error on fetching media for all: ${error}`); } }); }; public get isInactiveTime() { const timeToEpoch = (h = 0, m = 0) => new Date().setHours(h, m, 0, 0); return this.inactiveHours .map(rangeStr => ((start, end) => ({start, end}))( ...rangeStr.split('-', 2).map(timeStr => timeToEpoch(...timeStr.split(':', 2).map(Number))) as [number, number?] )) .some(range => (now => now >= range.start && now < range.end)(Date.now())); } public work = () => { const lock = this.lock; if (this.workInterval < 1) this.workInterval = 1; if (this.isInactiveTime || lock.feed.length === 0) { setTimeout(() => { this.workForAll(); setTimeout(this.work, this.workInterval * 200); }, this.workInterval * 1000 / lock.feed.length); return; } if (lock.workon >= lock.feed.length) lock.workon = 0; const currentFeed = lock.feed[lock.workon]; if (!lock.threads[currentFeed] || !lock.threads[currentFeed].subscribers || lock.threads[currentFeed].subscribers.length === 0) { logger.warn(`nobody subscribes thread ${currentFeed}, removing from feed`); delete lock.threads[currentFeed]; this.cache[parseLink(currentFeed).userName].pullOrder = 0; lock.feed.splice(lock.workon, 1); fs.writeFileSync(path.resolve(this.lockfile), JSON.stringify(lock)); this.work(); return; } logger.debug(`searching for new items from ${currentFeed} in cache`); const match = /https:\/\/www\.instagram\.com\/([^\/]+)/.exec(currentFeed); if (!match) { logger.error(`current feed "${currentFeed}" is invalid, please remove this feed manually`); lock.workon++; setTimeout(this.work, this.workInterval * 1000 / lock.feed.length); return; } const cachedFeed = this.cache[match[1]]; if (!cachedFeed) { setTimeout(this.work, this.workInterval * 1000 / lock.feed.length); return; } const newer = (item: CachedMediaItem) => BigNumOps.compare(item.pk, lock.threads[currentFeed].offset) > 0; const promise = Promise.resolve(Object.values(cachedFeed.stories) .filter(newer) .sort((i1, i2) => BigNumOps.compare(i2.pk, i1.pk)) .slice(-5) ); promise.then((mediaItems: CachedMediaItem[]) => { const currentThread = lock.threads[currentFeed]; if (!mediaItems || mediaItems.length === 0) return; const question = mediaItems.find(story => story.original.story_questions); const topOfFeed = question? question.pk : mediaItems[0].pk; const updateOffset = () => currentThread.offset = topOfFeed; if (currentThread.offset === '-1') { updateOffset(); return; } if (currentThread.offset === '0') mediaItems.splice(1); return this.workOnMedia(mediaItems.reverse(), this.sendStories(`thread ${currentFeed}`, ...currentThread.subscribers)) .then(updateOffset) .then(() => { if (question) { currentThread.subscribers.forEach(subscriber => { const username = cachedFeed.user.username; const author = `${cachedFeed.user.full_name} (@${username}) `; this.bot.sendTo(subscriber, `请注意,用户${author}已开启问答互动。需退订请回复:/igstory_unsub ${username}${ Object.keys(cachedFeed.stories).some(id => id > topOfFeed) ? `\n本次推送已截止于此条动态,下次推送在 ${Math.floor(this.workInterval * 1000 / lock.feed.length)} 秒后。` : ''}`); }); } }); }) .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); }); }; }