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 } 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 * as json from './json'; 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}; const igErrorIsAuthError = (error: IgClientError) => / 40[1-3]/.test(error.message) || error instanceof IgLoginRequiredError || error instanceof IgCookieNotFoundError; interface IWorkerOption { sessionLockfile: string; credentials: [string, string]; codeServicePort: number; proxyUrl: string; lock: ILock; lockfile: string; cache: ICache; cachefile: 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() .catch((err) => logger.error(err)) .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', })); } logger.error(err); }) .then(user => { logger.info(`successfully logged in as ${this.username}`); return 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 interface ICache { [userId: string]: { user: ReelsMediaFeedResponseItem['user']; stories: { [storyId: string]: CachedMediaItem; }; pullOrder: number; // one-based; -1: subscribed, awaiting shuffle; 0: not subscribed updated?: string; // Date.toString() }; } export default class { private client: IgApiClient; private lock: ILock; private lockfile: string; private cache: ICache; private cachefile: 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.cachefile = opt.cachefile; this.cache = opt.cache; 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: (userId: number | string) => void, retryAction: () => void, reply: (msg: string) => void }) => { const {action, retryAction, reply, rawUserName} = config; return this.queryUser(rawUserName) .then(userNameId => { const userId = userNameId.split(':')[1]; const lastUpdated = new Date((this.cache[userId] || {}).updated || 0).getTime(); const storyCount = lastUpdated && Object.keys(this.cache[userId].stories || {}).length; if (new Date().getTime() - lastUpdated > this.workInterval * 1000 && storyCount) return userId; return this.client.feed.reelsMedia({userIds: [userId]}).items() .then(storyItems => Promise.all(storyItems .filter(item => !(item.pk in this.cache[userId].stories)) .map(item => this.webshot( [{...item, user: this.cache[userId].user}], // guaranteed fresh cache entry (msgs: string, text: string, author: string) => this.cache[userId].stories[item.pk] = {pk: item.pk, msgs, text, author, original: item}, this.webshotDelay )) ).then(() => userId).finally(() => this.cache[userId].updated = Date())); }) .then(action) .catch((error: IgClientError & Partial) => { if (error instanceof IgExactUserNotFoundError) { reply(`找不到用户 ${rawUserName.replace(/^@?(.*)$/, '@$1')}。`); } else 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 (igErrorIsAuthError(error)) { 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: userId => { const userName = this.cache[userId].user.username; const storyItems = Object.values(this.cache[userId].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: userId => { if (!(storyId in this.cache[userId].stories)) return reply('此动态不存在或已过期。'); return this.workOnMedia([this.cache[userId].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 (count < 1) return reply('最大查看数量参数值应为正整数。'); const sender = this.sendStories(`instagram stories for ${rawUserName}`, receiver); workNow({ rawUserName, action: userId => { const userName = this.cache[userId].user.username; const storyItems = Object.values(this.cache[userId].stories) .sort((i1, i2) => -BigNumOps.compare(i2.pk, i1.pk)); // ascending! if (storyItems.length === 0) return reply(`当前用户 (@${userName}) 没有可用的 Instagram 限时动态。`); if (startIndex < 0) startIndex += storyItems.length; if (startIndex < 0 || 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, () => { const subscribedIds = this.lock.feed.map(feed => this.lock.threads[feed].id.toString()); for (const id in this.cache) { if (this.cache[id].pullOrder !== 0 && !subscribedIds.includes(id)) { logger.warn(`disabling pull job of unsubscribed user @${this.cache[id].user.username}`); this.cache[id].pullOrder = 0; } } const userIdCache = this.pullOrders; return (() => { if (Object.values(userIdCache).length !== userIdCache.length) { this.pullOrders = Arr.shuffle(userIdCache); return json.writeFile(path.resolve(this.cachefile), this.cache); } return Promise.resolve(true); })().then(() => { const timeout = Math.max(1000, this.workInterval * 1000 / this.lock.feed.length); setInterval(() => { this.pullOrders = Arr.shuffle(this.pullOrders); }, 21600000); setTimeout(this.workForAll, timeout); this.work(); }); } ); }; public queryUserObject = (userName: string) => this.client.user.searchExact(userName) .catch((error: IgClientError) => { if (igErrorIsAuthError(error)) { logger.warn('login required, logging in again...'); return this.session.login().then(() => this.client.user.searchExact(userName)); } else throw error; }); public queryUser = (rawUserName: string) => { const username = ScreenNameNormalizer.normalize(rawUserName).split(':')[0]; for (const {user} of Object.values(this.cache)) { if (user.username === username) return Promise.resolve(`${username}:${user.pk}`); } return this.queryUserObject(username) .then(({pk, username, full_name}) => { this.cache[pk] = {user: {pk, username, full_name}, stories: {}, pullOrder: 0}; return json.writeFile(path.resolve(this.cachefile), this.cache).then(res => { if (res) logger.info(`initialized cache item for user ${full_name} (@${username})`); return `${username}:${pk}`; }); }); }; private workOnMedia = ( mediaItems: CachedMediaItem[], sendMedia: (msg: string, text: string, author: string) => void ) => chainPromises(mediaItems.map(({msgs, text, author, original}) => { const findFilePath = (mediaMsg) => /=file:\/\/(.*?)]/.exec(mediaMsg)[1]; const filePath = findFilePath(msgs); return () => (fs.existsSync(filePath) ? Promise.resolve() : this.webshot.fetchBestCandidate(original).then(mediaMsg => { logger.warn(`media file missing, refetched media for ${author}/${original.code}`); return promisify(fs.rename)(findFilePath(mediaMsg), filePath); }) ).then(() => 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 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 = () => { const timeout = Math.max(1000, this.workInterval * 1000 / this.lock.feed.length); if (this.isInactiveTime) { setTimeout(this.workForAll, timeout); return; } 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...`); if (id in this.cache) { const item = this.cache[id]; if (item.pullOrder === 0) item.pullOrder = -1; return Promise.resolve(); } return promisify(setTimeout)((Math.random() * 2 + 1) * 5000).then(() => this.client.user.info(id).then(({pk, username, full_name}) => { this.cache[id] = {user: {pk, username, full_name}, stories: {}, pullOrder: -1}; return json.writeFile(path.resolve(this.cachefile), this.cache).then(() => logger.info(`initialized cache item for user ${full_name} (@${username})`) ); }) ); })) .then(() => { const userIdCache = Object.values(this.cache).some(item => item.pullOrder < 0) ? this.pullOrders = Arr.shuffle(Object.keys(this.cache)).map(Number) : this.pullOrders; return chainPromises( Arr.chunk(userIdCache.filter(v => Date.now() - new Date(this.cache[v].updated).getTime() > 3600000), 30).map(userIds => () => { logger.info(`pulling stories from users:${userIds.map(id => ` @${this.cache[id].user.username}`)}`); return this.client.feed.reelsMedia({userIds}).request() .then(({reels}) => chainPromises( Object.keys(reels).map(userId => this.cache[userId]).map(({user, stories}) => () => this.queryUserObject(user.username) .catch((error: IgClientError) => { if (error instanceof IgExactUserNotFoundError) { return this.client.user.info(user.pk) .then(({username, full_name}) => { user.username = username; user.full_name = full_name; }); } }).finally(() => Promise.all(reels[user.pk].items .filter(item => !(item.pk in stories)) .map(item => this.webshot( [{...item, user}], (msgs: string, text: string, author: string) => stories[item.pk] = {pk: item.pk, msgs, text, author, original: item}, this.webshotDelay )) ) ) ) )) .finally(() => json.writeFile(path.resolve(this.cachefile), this.cache).then(() => Object.values(this.lock.threads).forEach(thread => { if (userIds.includes(thread.id)) { thread.updatedAt = this.cache[thread.id].updated = Date(); } }) )) as unknown as Promise; }), (lp1, lp2) => () => lp1().then(() => promisify(setTimeout)(timeout).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 (igErrorIsAuthError(error)) { logger.warn('login required, logging in again...'); this.session.login().then(this.workForAll); } else { logger.error(`unhandled error on fetching media for all: ${error}`); } }) .then(() => { setTimeout(this.workForAll, this.workInterval * 1000); }); }; 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 => { const now = new Date().getTime(); return now >= range.start && now < range.end; }); } public work = () => { const lock = this.lock; const timeout = Math.max(1000, this.workInterval * 1000 / lock.feed.length); const nextTurn = () => { setTimeout(this.work, timeout); return; }; if (lock.feed.length === 0) return nextTurn(); 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] || {} as {pullOrder: number}).pullOrder = 0; return json.writeFile(path.resolve(this.cachefile), this.cache).then(() => { lock.feed.splice(lock.workon, 1); fs.writeFileSync(path.resolve(this.lockfile), JSON.stringify(lock)); return this.work(); }); } 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++; return nextTurn(); } const cachedFeed = this.cache[lock.threads[currentFeed].id]; if (!cachedFeed) return nextTurn(); 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 updateOffset = () => currentThread.offset = mediaItems[0].pk; if (currentThread.offset === '-1') { updateOffset(); return; } const questionIndex = mediaItems.findIndex(story => story.original.story_questions); if (questionIndex > 0) mediaItems.splice(0, questionIndex); if (currentThread.offset === '0') mediaItems.splice(1); return this.workOnMedia( mediaItems.slice(0).reverse(), this.sendStories(`thread ${currentFeed}`, ...currentThread.subscribers) ) .then(updateOffset) .then(() => { if (questionIndex > -1) { currentThread.subscribers.forEach(subscriber => { const username = cachedFeed.user.username; const author = `${cachedFeed.user.full_name} (@${username}) `; this.bot.sendTo(subscriber, `请注意,用户${author}已开启问答互动。需退订请回复:/igstory_unsub ${username}${ (questionIndex > 0) ? `\n本次推送已截止于此条动态,下次推送在 ${this.workInterval} 秒后。` : '' }` ); }); } }); }) .then(() => { lock.workon++; fs.writeFileSync(path.resolve(this.lockfile), JSON.stringify(lock)); nextTurn(); }); }; }