import type * as fs from 'fs' ; import { getPyTypedInfo } from './analyzer/pyTypedUtils' ; import { ExecutionEnvironment } from './common/configOptions' ; import { FileSystem , MkDirOptions } from './common/fileSystem' ; import { stubsSuffix } from './common/pathConsts' ; import { changeAnyExtension , combinePaths , ensureTrailingDirectorySeparator , isDirectory , tryStat , } from './common/pathUtils' ; import { ReadOnlyAugmentedFileSystem } from './readonlyAugmentedFileSystem' ; export class PyrightFileSystem extends ReadOnlyAugmentedFileSystem { private readonly _rootSearched = new Set < string > ( ) ; private readonly _partialStubPackagePaths = new Set < string > ( ) ; private readonly _conflictMap = new Map < string , string > ( ) ; private readonly _customUriMap = new Map < string , { uri : string ; closed : boolean ; hasPendingRequest : boolean ; } > ( ) ; constructor ( realFS ) { super ( realFS ) ; } override mkdirSync ( path , options ? ) { this . _realFS . mkdirSync ( path , options ) ; } override chdir ( path ) { this . _realFS . chdir ( path ) ; } override writeFileSync ( path , data , encoding ) { this . _realFS . writeFileSync ( this . _getOriginalPath ( path ) , data , encoding ) ; } override unlinkSync ( path ) { this . _realFS . unlinkSync ( this . _getOriginalPath ( path ) ) ; } override createWriteStream ( path ) { return this . _realFS . createWriteStream ( this . _getOriginalPath ( path ) ) ; } override copyFileSync ( src , dst ) { this . _realFS . copyFileSync ( this . _getOriginalPath ( src ) , this . _getOriginalPath ( dst ) ) ; } override getUri ( originalPath ) { const entry = this . _customUriMap . get ( this . getMappedFilePath ( originalPath ) ) ; if ( entry ) { return entry . uri ; } return this . _realFS . getUri ( originalPath ) ; } hasUriMapEntry ( uriString , mappedPath ) { const entry = this . _customUriMap . get ( mappedPath ) ; if ( ! entry || entry . uri !== uriString ) { return false ; } return true ; } addUriMap ( uriString , mappedPath ) { const entry = this . _customUriMap . get ( mappedPath ) ; if ( ! entry ) { this . _customUriMap . set ( mappedPath , { uri : uriString , closed : false , hasPendingRequest : false } ) ; return true ; } if ( entry . uri !== uriString ) { return false ; } entry . closed = false ; return true ; } removeUriMap ( uriString , mappedPath ) { const entry = this . _customUriMap . get ( mappedPath ) ; if ( ! entry || entry . uri !== uriString ) { return false ; } if ( entry . hasPendingRequest ) { entry . closed = true ; return true ; } this . _customUriMap . delete ( mappedPath ) ; return true ; } pendingRequest ( mappedPath , hasPendingRequest ) { const entry = this . _customUriMap . get ( mappedPath ) ; if ( ! entry ) { return ; } if ( ! hasPendingRequest && entry . closed ) { this . _customUriMap . delete ( mappedPath ) ; return ; } entry . hasPendingRequest = hasPendingRequest ; } isPartialStubPackagesScanned ( execEnv ) { return this . isPathScanned ( execEnv . root ?? '' ) ; } isPathScanned ( path ) { return this . _rootSearched . has ( path ) ; } processPartialStubPackages ( paths , roots ) { for ( const path of paths ) { this . _rootSearched . add ( path ) ; if ( ! this . _realFS . existsSync ( path ) || ! isDirectory ( this . _realFS , path ) ) { continue ; } for ( const entry of this . _realFS . readdirEntriesSync ( path ) ) { const partialStubPackagePath = combinePaths ( path , entry . name ) ; const isDirectory = ! entry . isSymbolicLink ( ) ? entry . isDirectory ( ) : ! ! tryStat ( this . _realFS , partialStubPackagePath ) ?. isDirectory ( ) ; if ( ! isDirectory || ! entry . name . endsWith ( stubsSuffix ) ) { continue ; } const pyTypedInfo = getPyTypedInfo ( this . _realFS , partialStubPackagePath ) ; if ( ! pyTypedInfo || ! pyTypedInfo . isPartiallyTyped ) { continue ; } this . _partialStubPackagePaths . add ( partialStubPackagePath ) ; let partialStubs ; const packageName = entry . name . substr ( 0 , entry . name . length - stubsSuffix . length ) ; for ( const root of roots ) { const packagePath = combinePaths ( root , packageName ) ; try { const stat = tryStat ( this . _realFS , packagePath ) ; if ( ! stat ?. isDirectory ( ) ) { continue ; } const packagePyTyped = getPyTypedInfo ( this . _realFS , packagePath ) ; if ( packagePyTyped && ! packagePyTyped . isPartiallyTyped ) { continue ; } partialStubs = partialStubs ?? this . _getRelativePathPartialStubs ( partialStubPackagePath ) ; for ( const partialStub of partialStubs ) { const originalPyiFile = combinePaths ( partialStubPackagePath , partialStub ) ; const mappedPyiFile = combinePaths ( packagePath , partialStub ) ; if ( this . existsSync ( mappedPyiFile ) ) { const mappedPyFile = changeAnyExtension ( mappedPyiFile , 'py' ) ; const tmpPyFile = changeAnyExtension ( combinePaths ( this . tmpdir ( ) , 'conflictFiles' , packageName , partialStub ) , 'py' ) ; if ( ! this . _conflictMap . has ( mappedPyiFile ) && ! this . existsSync ( mappedPyFile ) && ! this . existsSync ( tmpPyFile ) ) { this . _recordMovedEntry ( tmpPyFile , originalPyiFile , false ) ; this . _conflictMap . set ( mappedPyiFile , tmpPyFile ) ; } continue ; } this . _recordMovedEntry ( mappedPyiFile , originalPyiFile ) ; } } catch { } } } } } clearPartialStubs ( ) { super . _clear ( ) ; this . _rootSearched . clear ( ) ; this . _partialStubPackagePaths . clear ( ) ; this . _conflictMap . clear ( ) ; } getConflictedFile ( filepath ) { return this . _conflictMap . get ( filepath ) ; } private _getRelativePathPartialStubs ( path ) { const paths = [ ] ; const partialStubPathLength = ensureTrailingDirectorySeparator ( path ) . length ; const searchAllStubs = ( path ) => { for ( const entry of this . _realFS . readdirEntriesSync ( path ) ) { const filePath = combinePaths ( path , entry . name ) ; let isDirectory = entry . isDirectory ( ) ; let isFile = entry . isFile ( ) ; if ( entry . isSymbolicLink ( ) ) { const stat = tryStat ( this . _realFS , filePath ) ; if ( stat ) { isDirectory = stat . isDirectory ( ) ; isFile = stat . isFile ( ) ; } } if ( isDirectory ) { searchAllStubs ( filePath ) ; } if ( isFile && entry . name . endsWith ( '.pyi' ) ) { const relative = filePath . substring ( partialStubPathLength ) ; if ( relative ) { paths . push ( relative ) ; } } } } ; searchAllStubs ( path ) ; return paths ; } protected override _isMovedEntry ( path ) { return this . _partialStubPackagePaths . has ( path ) || super . _isMovedEntry ( path ) ; } }